Agentfield
AgentField is an open-source control plane that turns AI agents written in Python, Go, or TypeScript into production-ready REST APIs. It handles fan-out, queuing, retries, and observability, so one request can scale to thousands of agents. It is for developers who need to build and deploy multi-agent systems at scale.
✨ Key features
- Turns plain functions into REST endpoints automatically
- Fan-out to thousands of agents with queuing and retries
- Structured AI output with Pydantic/Zod schemas
- Human-in-the-loop approvals with pause and resume
- Cross-agent calls with service discovery and tracing
- Multi-language SDKs: Python, Go, TypeScript
🎯 Use cases
- Build a claims processor with risk scoring and human approval
- Create a deep research engine that fans out to thousands of agents
- Run a security auditor with 250 coordinated agents per audit
- Expose agents as APIs for frontends, backends, or cron jobs
📦 Installation
🧰 Requirements: Requires Python, Go, or TypeScript runtime; Docker for control plane deployment; API keys for LLM providers (e.g., Anthropic).
curl -fsSL https://agentfield.ai/install.sh | bash
The installer also drops the aforge coding harness beside af in ~/.agentfield/bin, so harness-backed agents work out of the box; skip it with --no-aforge.
On macOS the installer also registers the control plane to start at login (under launchd) and adds a menu-bar icon. Stop it with af service stop or the menu-bar icon — a plain kill looks like a crash and it restarts. af service status shows health and in-flight work; install with --no-tray to skip this entirely.
🚀 Usage
import asyncio
from agentfield import Agent, AIConfig
from pydantic import BaseModel
app = Agent(
node_id="researcher",
version="1.0.0",
ai_config=AIConfig(model="anthropic/claude-sonnet-4-20250514"),
)
class SubQuestions(BaseModel):
questions: list[str]
@app.reasoner(tags=["research"])
async def research(question: str, depth: int = 0, model: str | None = None) -> dict:
if depth >= 3:
answer = await app.ai(system="Answer directly and concisely.", user=question, model=model)
return {"question": question, "answer": answer}
plan = await app.ai(
system="Break this into 3-5 independent sub-questions.",
user=question, schema=SubQuestions, model=model,
)
branches = await asyncio.gather(*[
app.call(f"{app.node_id}.research", question=q, depth=depth + 1, model=model)
for q in plan.questions
])
synthesis = await app.ai(system="Synthesize these findings.", user=str(branches), model=model)
return {"question": question, "answer": synthesis, "branches": branches}
app.run()
⚠️ Good to know
Routing overhead is roughly 100-200ms per cross-agent hop, so keep hops coarse when latency is tight.
❓ FAQ
What languages are supported?
AgentField supports Python, Go, and TypeScript SDKs.
How does AgentField handle scaling?
The control plane is a stateless Go service that scales horizontally behind a load balancer, with bounded in-process queues and backpressure.
Can I integrate with existing frameworks like LangChain?
Yes, a reasoner is a plain function, so existing LangGraph or CrewAI code can run inside one.
How do I get human approval in a workflow?
Use app.pause() to suspend execution, notify via webhook, and resume when approved.
📊 Repository
🤖 Overview, features, install steps and FAQ were generated from the project's README on Sep 4, 2026. Always check the original source before running commands.