Python SDK

Agent Constructor

from agentfield import Agent, AIConfig, HarnessConfig, MemoryConfig

app = Agent(
    node_id="my-agent",                           # Required — unique agent identifier
    agentfield_server="http://localhost:8080",     # Control plane URL (default: env or localhost:8080)
    version="1.0.0",                               # Agent version string
    description="My agent description",            # Human-readable description
    tags=["nlp", "production"],                    # Organizational tags
    author={"name": "Team", "email": "a@b.com"},  # Author metadata
    ai_config=AIConfig(...),                       # LLM configuration
    harness_config=HarnessConfig(...),             # Coding agent configuration (defaults to aforge)
    memory_config=MemoryConfig(...),               # Memory behavior configuration
    dev_mode=False,                                # Enable verbose debug logging
    async_config=None,                             # AsyncConfig for execution tuning
    callback_url=None,                             # Explicit callback URL for control plane
    auto_register=True,                            # Auto-register on first invocation
    vc_enabled=True,                               # Generate verifiable credentials
    api_key=None,                                  # API key for control plane auth
    enable_mcp=False,                              # Enable Model Context Protocol servers
    enable_did=True,                               # Enable decentralized identity
    local_verification=False,                      # Enable local DID signature verification
    verification_refresh_interval=300,             # DID verification refresh interval (seconds)
)

The Agent class extends FastAPI, so you can use all FastAPI features (middleware, dependency injection, lifespan events) alongside AgentField capabilities.

Environment Variables

Complete Method Reference

Core Methods

Properties

app.ai() — LLM Calls

The universal AI method supports text, images, audio, and file inputs with intelligent type detection.

async def ai(
    *args: Any,                                    # Flexible inputs — text, URLs, bytes, dicts
    system: Optional[str] = None,                  # System prompt
    user: Optional[str] = None,                    # User message (alternative to positional args)
    schema: Optional[Type[BaseModel]] = None,      # Pydantic model for structured output
    model: Optional[str] = None,                   # Override model (e.g., "gpt-4o", "claude-3")
    temperature: Optional[float] = None,           # Creativity (0.0-2.0)
    max_tokens: Optional[int] = None,              # Maximum response length
    stream: Optional[bool] = None,                 # Enable streaming
    response_format: Optional[str | Dict] = None,  # "auto", "json", or "text"
    context: Optional[Dict] = None,                # Additional context for the LLM
    memory_scope: Optional[List[str]] = None,      # Memory scopes to inject
    tools: Optional[...] = None,                   # Tool definitions for tool calling
    max_turns: Optional[int] = None,               # Max LLM turns in tool-call loop (default: 10)
    max_tool_calls: Optional[int] = None,          # Max total tool calls (default: 25)
    **kwargs,                                      # Additional provider-specific parameters
) -> Any

Examples

# Simple text
response = await app.ai("Summarize this document.")

# System + user prompts
response = await app.ai(
    system="You are a helpful assistant.",
    user="What is the capital of France?",
)

# Structured output with Pydantic
class Sentiment(BaseModel):
    sentiment: str
    confidence: float

result = await app.ai(
    "Analyze sentiment of: I love this!",
    schema=Sentiment,
)
print(result.sentiment)      # "positive"
print(result.confidence)     # 0.95

# Multimodal — image + text
response = await app.ai(
    "Describe this image:",
    "https://example.com/photo.jpg",
)

# Tool calling with auto-discovery
response = await app.ai(
    system="You are a helpful assistant with access to tools.",
    user="What agents are available?",
    tools="discover",
)

Tool Calling

The tools parameter accepts multiple formats:

from agentfield import ToolCallConfig

result = await app.ai(
    user="Process this claim",
    tools=ToolCallConfig(
        max_turns=5,
        max_tool_calls=10,
        tags=["insurance"],
        agent_ids=["claims-processor"],
    ),
)

app.harness() — Harness orchestration

Dispatch tasks to a coding agent with tool access, structured output, and runtime limits.

With no provider, the call runs AForge — AgentField's native harness, installed alongside the af binary, so there is nothing extra to set up beyond OPENROUTER_API_KEY. Pass provider= to drive Claude Code, Codex, Gemini CLI, OpenCode, or Grok (Python SDK only) instead. Precedence: an explicit provider argument, then the agent's HarnessConfig, then AGENTFIELD_HARNESS_PROVIDER, then "aforge".

async def harness(
    prompt: str,                                    # Task description
    *,
    schema: Any = None,                             # Pydantic model for structured output
    provider: Optional[str] = None,                 # None -> "aforge"; else "claude-code", "codex", "gemini", "opencode", "grok"
    model: Optional[str] = None,                    # None -> the provider's own default model
    max_turns: Optional[int] = None,                # Maximum agent iterations
    max_budget_usd: Optional[float] = None,         # Cost cap in USD (claude-code only)
    tools: Optional[List[str]] = None,              # Allowed tools (claude-code only)
    permission_mode: Optional[str] = None,          # "plan", "auto", or None (claude-code, codex, gemini)
    system_prompt: Optional[str] = None,            # System prompt for the agent
    env: Optional[Dict[str, str]] = None,           # Environment variables
    cwd: Optional[str] = None,                      # Working directory
    **kwargs,
) -> HarnessResult

HarnessResult

Example

from pydantic import BaseModel

class CodeReview(BaseModel):
    issues: list[str]
    suggestions: list[str]
    score: int

# No provider, no model — this runs on AForge, the default harness.
result = await app.harness(
    "Review the Python code in ./src for security issues",
    schema=CodeReview,
    max_turns=20,
    cwd="/path/to/project",
)

print(result.parsed.score)       # 85
print(result.cost_usd)           # 0.23

app.call() — Cross-Agent Communication

Route calls to other agents through the control plane. Always returns dict (like a REST API).

async def call(
    target: str,       # "node_id.function_name"
    *args,             # Positional arguments (auto-mapped to parameter names)
    **kwargs,          # Keyword arguments
) -> dict

Examples

# Keyword arguments (recommended)
result = await app.call(
    "sentiment-agent.analyze",
    message="I love this product!",
    customer_id="cust_123",
)

# Positional arguments
result = await app.call(
    "notification-agent.send_email",
    "user@example.com",       # to
    "Welcome!",               # subject
    body="Thanks for joining.",
)

app.pause() — Human-in-the-Loop

Suspend execution until a human approves, rejects, or requests changes.

async def pause(
    approval_request_id: str,            # ID of the approval request
    approval_request_url: str = "",      # URL where humans review the request
    expires_in_hours: int = 72,          # Expiry time
    timeout: Optional[float] = None,     # Max seconds to wait
    execution_id: Optional[str] = None,  # Override current execution ID
) -> ApprovalResult

ApprovalResult

Decorators

@app.reasoner()

Register an AI-powered function.

@app.reasoner(
    path=None,                             # Custom endpoint path
    name=None,                             # Explicit registration ID
    tags=None,                             # Organizational tags
    vc_enabled=None,                       # Override VC generation
    require_realtime_validation=False,     # Require real-time VC validation
)

@app.skill()

Register a deterministic function.

@app.skill(
    tags=None,                             # Organizational tags
    path=None,                             # Custom endpoint path
    name=None,                             # Explicit registration ID
    vc_enabled=None,                       # Override VC generation
    require_realtime_validation=False,     # Require real-time VC validation
)

Both decorators automatically generate input/output schemas from type hints, create FastAPI endpoints with validation, integrate with workflow tracking, and enable cross-agent communication via app.call().

@app.session()

Register a realtime or multimodal session entrypoint. Sessions start through the AgentField control plane, while session tools still route into normal reasoners and workflow DAGs.

@app.session(
    "voice",
    provider="openai",
    transport="webrtc",
    model="gpt-realtime-2",
    modalities=["audio", "text"],
    voice="marin",
    tools=["voice-support-af.resolve_voice_turn"],
    tags=["support:voice", "pii:limited"],
)
async def voice(session):
    turn = await session.input()
    result = await session.call("voice-support-af.resolve_voice_turn", turn=turn)
    await session.say(result["spoken_response"])

Provider and transport are explicit. AgentField validates the combination and does not infer or switch providers.

tools=[...] is a provider/client-visible allowlist, not a requirement for the handler to call reasoners. The handler can always orchestrate with session.call(...); tools exposes selected AgentField targets for autonomous realtime tool calls during the live session.

tags=[...] proposes access-control tags for the session ingress. Approve them like reasoner and skill tags, then use policies to control who can start the live session.

Memory

The app.memory property provides a MemoryInterface scoped to the current execution context.

Memory Scopes

MemoryInterface Methods

# Basic key-value operations
await app.memory.set(key: str, data: Any) -> None
await app.memory.get(key: str, default: Any = None) -> Any
await app.memory.exists(key: str) -> bool
await app.memory.delete(key: str) -> None

# Vector operations
await app.memory.set_vector(key, embedding, metadata=None)
await app.memory.delete_vector(key)
await app.memory.similarity_search(query_embedding, top_k=10, filters=None)

# Scoped access
app.memory.session(session_id) -> ScopedMemoryClient
app.memory.actor(actor_id) -> ScopedMemoryClient
app.memory.workflow(workflow_id) -> ScopedMemoryClient

# Reactive events
@app.memory.on_change("user.*")
async def on_user_change(event):
    print(f"Key {event.key} changed: {event.data}")

Configuration Classes

AIConfig

from agentfield import AIConfig

config = AIConfig(
    model="gpt-4o",                          # Default LLM model
    temperature=None,                        # 0.0-2.0
    max_tokens=None,                         # Max response length
    fallback_models=[],                      # Fallback model list
    enable_rate_limit_retry=True,            # Auto-retry on rate limits
    max_cost_per_call=None,                  # Max cost per call in USD
    daily_budget=None,                       # Daily budget in USD
    auto_inject_memory=[],                   # Memory scopes to auto-inject
    litellm_params={},                       # Additional LiteLLM parameters
)

HarnessConfig

from agentfield import HarnessConfig

config = HarnessConfig(
    # provider defaults to "aforge" — omit it for the zero-setup default.
    # Override with "claude-code", "codex", "gemini", "opencode", or "grok".
    model=None,                              # None -> the provider's own default model
    max_turns=30,                            # Maximum agent iterations
    max_budget_usd=None,                     # Cost cap in USD (claude-code only)
    max_retries=3,                           # Retry attempts
    tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"],  # claude-code only
    permission_mode=None,                    # "plan", "auto", or None (claude-code, codex, gemini)
)

MemoryConfig

from agentfield import MemoryConfig

config = MemoryConfig(
    auto_inject=["workflow", "session"],     # Required — memory scopes to auto-inject
    memory_retention="session",              # Required — retention policy
    cache_results=True,                      # Required — cache memory results locally
)

AgentRouter

Organize reasoners and skills into modular routers, similar to FastAPI's APIRouter.

from agentfield import Agent, AgentRouter

user_router = AgentRouter(prefix="/users", tags=["users"])

@user_router.reasoner()
async def analyze_behavior(user_id: str) -> dict:
    ...

@user_router.skill()
async def get_profile(user_id: str) -> dict:
    ...

app = Agent(node_id="my-agent")
app.include_router(user_router)

Exceptions

from agentfield import (
    AgentFieldError,           # Base exception
    AgentFieldClientError,     # Control plane request failures
    ExecutionTimeoutError,     # Execution timeout
    MemoryAccessError,         # Memory backend failures
    RegistrationError,         # Agent registration failures
    ValidationError,           # Input validation failures
)

Complete Example

from agentfield import Agent, AIConfig, HarnessConfig
from pydantic import BaseModel

app = Agent(
    node_id="claims-processor",
    version="2.1.0",
    ai_config=AIConfig(model="anthropic/claude-sonnet-4-20250514"),
    tags=["insurance", "critical"],
)

class Decision(BaseModel):
    action: str         # "approve", "deny", "escalate"
    confidence: float
    reasoning: str

@app.reasoner(tags=["insurance", "critical"])
async def evaluate_claim(claim: dict) -> dict:
    await app.memory.set("current_claim", claim)

    decision = await app.ai(
        system="Insurance claims adjuster. Evaluate and decide.",
        user=f"Claim #{claim['id']}: {claim['description']}",
        schema=Decision,
    )

    app.note(f"Decision: {decision.action} ({decision.confidence})", ["audit"])

    if decision.confidence < 0.85:
        approval = await app.pause(
            approval_request_id=f"claim-{claim['id']}",
            approval_request_url=f"https://internal.acme.com/approvals/{claim['id']}",
            expires_in_hours=48,
        )
        if not approval.approved:
            return {"action": "rejected", "feedback": approval.feedback}

    await app.call("notifier.send_decision",
        claim_id=claim["id"],
        decision=decision.model_dump(),
    )

    return decision.model_dump()

app.run()