For Developers

Write a function.Get an agent.

Async execution. No timeout limits. Signed webhooks. Cross-agent discovery. Human-in-the-loop. Scales to thousands. Python, Go, or TypeScript.

Star on GitHub...Quickstart

$curl -sSf https://agentfield.ai/get | sh

main.py

Python

1from agentfield import Agent, AIConfig
2from pydantic import BaseModel
 
3app = Agent(
4node_id="claims-processor",
5version="2.1.0",# Canary deploys, A/B testing, blue-green rollouts
6ai_config=AIConfig(model="anthropic/claude-sonnet-4-20250514"),
7)
 
8class Decision(BaseModel):
9action: str# "approve", "deny", "escalate"
10confidence: float
11reasoning: str
 
12@app.reasoner(tags=["insurance", "critical"])discoverable
13async def evaluate_claim(claim: dict) -> dict:
 
14# Structured AI judgment - returns typed Pydantic output
15decision = await app.ai(structured AI
16system="Insurance claims adjuster. Evaluate and decide.",
17user=f"Claim #{claim['id']}: {claim['description']}",
18schema=Decision,
19)
 
20if decision.confidence < 0.85:
21# Human approval - suspends, notifies, resumes when approved
22await app.pause(human-in-the-loop
23approval_request_id=f"claim-{claim['id']}",
24expires_in_hours=48,
25)
 
26# Route to the next agent - traced through the control plane
27await app.call("notifier.send_decision", input={agent mesh
28"claim_id": claim["id"],
29"decision": decision.model_dump(),
30})
 
31return decision.model_dump()
 
32app.serve()one line to production
33# Exposes: POST /api/v1/execute/claims-processor.evaluate_claim
34# Auto-registers, cryptographic identity, tamper-proof audit trail.

main.py

Python

1from agentfield import Agent, AIConfig
2from pydantic import BaseModel
 
3app = Agent(
4node_id="claims-processor",
5version="2.1.0",# Canary deploys, A/B testing, blue-green rollouts
6ai_config=AIConfig(model="anthropic/claude-sonnet-4-20250514"),
7)
 
8class Decision(BaseModel):
9action: str# "approve", "deny", "escalate"
10confidence: float
11reasoning: str
 
12@app.reasoner(tags=["insurance", "critical"])discoverable
13async def evaluate_claim(claim: dict) -> dict:
 
14# Structured AI judgment - returns typed Pydantic output
15decision = await app.ai(structured AI
16system="Insurance claims adjuster. Evaluate and decide.",
17user=f"Claim #{claim['id']}: {claim['description']}",
18schema=Decision,
19)
 
20if decision.confidence < 0.85:
21# Human approval - suspends, notifies, resumes when approved
22await app.pause(human-in-the-loop
23approval_request_id=f"claim-{claim['id']}",
24expires_in_hours=48,
25)
 
26# Route to the next agent - traced through the control plane
27await app.call("notifier.send_decision", input={agent mesh
28"claim_id": claim["id"],
29"decision": decision.model_dump(),
30})
 
31return decision.model_dump()
 
32app.serve()one line to production
33# Exposes: POST /api/v1/execute/claims-processor.evaluate_claim
34# Auto-registers, cryptographic identity, tamper-proof audit trail.

The Code

Your agent. Everything it needs.

A claims-processor that reasons about input, pauses for human approval, calls other agents, and runs async with signed webhooks. Click any tab to explore.

Decorator to API Endpoint

@app.reasoner() turns any function into a production REST endpoint. Add app.ai(schema=Decision) and the LLM returns typed Pydantic output. No Flask, no FastAPI, no string parsing.

Single entry pointPOST /api/v1/execute/{node_id}.{function}

@app.reasoner()evaluate_claim/execute/claims-processor.evaluate_claim

@app.reasoner()route_claim/execute/claims-processor.route_claim

app.ai(schema=Decision)

action: "approve"

confidence: 0.92

reasoning: "Claim matches..."

Decorate, serve, every function is callable via REST with typed output

claims-processor

1from agentfield import Agent, AIConfig
2from pydantic import BaseModel
3 
4app = Agent(
5    node_id="claims-processor",
6    version="2.1.0", # Release version
7    ai_config=AIConfig(model="gpt-4o"),
8)
9 
10class Decision(BaseModel):
11    action: str  # "approve", "deny", "escalate"
12    confidence: float
13    reasoning: str
14 
15@app.reasoner(tags=["insurance", "critical"])
16async def evaluate_claim(claim: dict) -> dict:
17 
18    # Structured AI judgment
19    response = await app.ai(f"Claim #{claim['id']}: {claim['description']}",
20        system="Insurance claims adjuster. Evaluate and decide.",
21        schema=Decision,
22    )
23    decision = response
24 
25    if decision.confidence < 0.85:
26        # Human approval. Suspends, notifies, resumes when approved.
27        await app.pause(
28            approval_request_id=f"claim-{claim['id']}",
29            expires_in_hours=48,
30        )
31 
32    # Route to the next agent
33    await app.call("notifier.send_decision",
34        claim_id=claim["id"],
35        decision=decision.model_dump(),
36    )
37 
38    return decision.model_dump()
39 
40@app.reasoner()
41async def route_claim(claim_id: str, category: str):
42    """Discover the best agent at runtime."""
43    discovery = await app.discover(
44        tags=["resolve", category],
45        format="compact",
46    )
47 
48    return await app.call(
49        discovery.compact.reasoners[0].target,
50        claim_id=claim_id
51    )
52 
53# Harness: dispatch multi-turn task to a coding agent
54result = await app.harness(
55    "Audit the claims pipeline for edge cases",
56    provider="claude-code",
57    max_budget_usd=5.0,
58)
59 
60# Memory: globally shared state
61await app.memory.global_scope.set("last_claim", claim_id)
62 
63app.serve(port=8001)

01Decorator to API Endpoint

@app.reasoner() turns any function into a production REST endpoint. Add app.ai(schema=Decision) and the LLM returns typed Pydantic output. No Flask, no FastAPI, no string parsing.

Single entry pointPOST /api/v1/execute/{node_id}.{function}

@app.reasoner()evaluate_claim/execute/claims-processor.evaluate_claim

@app.reasoner()route_claim/execute/claims-processor.route_claim

app.ai(schema=Decision)

action: "approve"

confidence: 0.92

reasoning: "Claim matches..."

Decorate, serve, every function is callable via REST with typed output

Infrastructure

What You Don't Build

Built into the control plane. Not your problem.

Zero-Config Memory

KV + vector search, four scopes (global, agent, session, run). No Redis, no Pinecone. Just use it.

results = await app.memory.search(
    query="deployment failures",
    scope="workflow",
    top_k=5
)

ReplacesRedis, Pinecone, pgvector, Weaviate setup

Auto Observability

Workflow DAGs, Prometheus metrics, and execution timeline. Generated automatically. Zero instrumentation code.

app.note(
    "## Risk Assessment\nScore: 0.82",
    tags=["audit", "risk"]
)
# appears in workflow timeline

ReplacesOpenTelemetry setup, Datadog instrumentation, manual tracing

Reactive Memory Events

Trigger agent logic when memory changes. Pattern-matched keys, real-time callbacks. No Kafka, no message bus.

@app.memory.on_change("order_*")
async def handle_order(key, value):
    await app.call(
        "fulfillment.process", order=value
    )

ReplacesKafka consumers, Redis pub/sub, custom event handlers

Webhooks & Streaming

HMAC-signed webhooks, SSE streaming, and polling. All built in. Pick the pattern that fits.

result = await app.call(
    "report.generate",
    mode="async",
    webhook="https://api.co/hook",
    stream=True
)

ReplacesCustom webhook infra, SSE plumbing, WebSocket servers

AI Tool Calling

Set tools="discover" and the LLM finds, selects, and orchestrates agents automatically. No manual function schemas.

result = await app.ai(
    system="You are a research analyst.",
    user=query,
    tools="discover"
)

ReplacesLangChain tool wrappers, manual function schemas

Fleet Management

Connector API for remote agent management. Version traffic control, bearer token auth, air-gapped deployment.

# Connector API
GET  /connector/reasoners
POST /connector/.../weight
# Bearer token auth
# Outbound WebSocket only

ReplacesCustom fleet management, manual deployment scripts

Policy Engine

Tag-based ALLOW/DENY rules on caller-to-target. Enforced by infrastructure, not prompts. Cryptographic verification.

# Only "finance" tagged agents can call this
policy = Policy(
    action="ALLOW",
    caller_tags=["finance"],
)

ReplacesCustom RBAC, manual access control, prompt-based guardrails

Examples

Start From an ExampleOpen Source

Production architectures you can clone. Security auditing, deep research, autonomous engineering, reactive data enrichment.

SEC-AF

250 coordinated agents trace every vulnerability source-to-sink and adversarially verify each finding. Confirmed exploits, not pattern flags.

250 agents · HUNT→PROVE · Streaming pipeline

examples/sec-af/

Deep Research

Recursive research backend. Spawns parallel agents, evaluates quality, generates deeper agents, and recurses. 10,000+ agents per query.

10,000+ agents · Fan-out→Filter→Recurse

examples/af-deep-research/

SWE-AF

Hundreds of coordinated agents that plan, build, test, and ship. Factory control loops, topological task ordering, and parallel git worktrees.

Multi-agent · Topological DAG · Parallel worktrees

examples/af-swe/

Reactive MongoDB

Atlas Triggers + agent reasoning. Documents arrive raw and leave enriched with risk scores, pattern detection, and evidence chains.

Multi-agent · Reactive enrichment · MongoDB

examples/reactive-atlas/

10+ open-source examplesSecurity, research, healthcare, finance, voice, and more. Clone any repo and start building.

View all examples

Async execution. Human approvals.
Agent discovery. One install.

Star the repo. Try the quickstart. Build something that runs in production.

Star on GitHub...Join Discord400+

PythonGoTypeScript

Read the docs

Write a function.Get an agent.

Your agent. Everything it needs.

Decorator to API Endpoint

01Decorator to API Endpoint

What You Don't Build

Zero-Config Memory

Auto Observability

Reactive Memory Events

Webhooks & Streaming

AI Tool Calling

Fleet Management

Policy Engine

Start From an ExampleOpen Source

SEC-AF

Deep Research

SWE-AF

Reactive MongoDB

Async execution. Human approvals.Agent discovery. One install.

Your agent. Everything it needs.

Decorator to API Endpoint

01Decorator to API Endpoint

What You Don't Build

Zero-Config Memory

Auto Observability

Reactive Memory Events

Webhooks & Streaming

AI Tool Calling

Fleet Management

Policy Engine

Start From an ExampleOpen Source

SEC-AF

Deep Research

SWE-AF

Reactive MongoDB

Async execution. Human approvals.Agent discovery. One install.

Async execution. Human approvals.
Agent discovery. One install.

Async execution. Human approvals.
Agent discovery. One install.