AgentFieldbuild

Cross-agent calls

Call reasoners and skills on other agents through the control plane execution gateway.

Call any reasoner or skill on any agent -- one line, full context propagation, automatic DAG tracking.

Building one agent is easy. Building five agents owned by different teams is where the infrastructure work starts: service discovery, routing, correlation IDs, workflow tracing, and failure debugging.

app.call() is the abstraction that collapses that distributed-systems work. The caller only needs node_id.function_name. AgentField handles the routing, workflow propagation, and trace construction behind it.

@app.reasoner()
async def triage_ticket(ticket_id: str, message: str, customer_id: str) -> dict:
    sentiment = await app.call("sentiment-agent.analyze_text", text=message)
    history = await app.call("customer-history.recent_issues", customer_id=customer_id)

    if sentiment["urgency"] == "high" or history["issue_count"] > 3:
        escalation = await app.call(
            "escalation-agent.create_case",
            ticket_id=ticket_id,
            reason="high urgency or repeated incidents",
        )
        await app.call("notifications.send_alert", case_id=escalation["case_id"])
        app.note(f"Escalated ticket {ticket_id} to {escalation['case_id']}", ["triage", "escalation"])
        return {"status": "escalated", "case_id": escalation["case_id"]}

    return {"status": "queued", "sentiment": sentiment, "history": history}

What just happened

The caller referenced services by logical target, not URL
Workflow context flowed through every child call automatically
Every child execution became a node in the same DAG
The note on escalation was attached to the current execution timeline

Example workflow proof:

{
  "target": "support-triage.triage_ticket",
  "status": "completed",
  "children": [
    { "target": "sentiment-agent.analyze_text", "status": "completed" },
    { "target": "customer-history.recent_issues", "status": "completed" },
    { "target": "escalation-agent.create_case", "status": "completed" },
    { "target": "notifications.send_alert", "status": "completed" }
  ]
}

What you get

Location-transparent calls -- address agents by "node_id.function_name", not URLs.
Automatic context propagation -- workflow ID, session ID, actor ID, and parent execution ID flow through every call via HTTP headers.
Workflow DAG building -- every cross-agent call creates a parent-child edge in the execution graph, visible in the dashboard.
Local call optimization -- when the target is on the same agent, the call bypasses the network entirely.
Consistent return type -- always returns a dict/object, like calling a REST API.

Full SDK examples

from agentfield import Agent

app = Agent(node_id="orchestrator")

@app.reasoner()
async def process_order(order: dict) -> dict:
    # Call a reasoner on another agent
    sentiment = await app.call(
        "sentiment-agent.analyze",
        text=order["customer_message"],
    )

    # Call a skill on another agent
    notification = await app.call(
        "notifier.send_email",
        to=order["customer_email"],
        subject="Order Received",
        body=f"Sentiment: {sentiment['label']}",
    )

    return {
        "order_id": order["id"],
        "sentiment": sentiment,
        "notified": True,
    }

Context propagation

Every cross-agent call automatically forwards execution context through HTTP headers. This is how AgentField builds workflow DAGs and maintains traceability across agent boundaries.

Header	Purpose
`X-Run-ID`	Groups all executions in a single top-level invocation
`X-Workflow-ID`	Identifies the workflow this execution belongs to
`X-Parent-Execution-ID`	Links child execution to its parent
`X-Session-ID`	Carries the session context across agents
`X-Actor-ID`	Carries the actor/user identity across agents
`X-Caller-DID`	Decentralized identifier of the calling agent

You never set these headers manually. The SDK reads them from the current execution context and forwards them on every call(). The receiving agent parses them and makes them available via ExecutionContext.

Accessing Execution Context

@app.reasoner()
async def handler(text: str, execution_context=None) -> dict:
    # execution_context is injected automatically
    print(execution_context.run_id)
    print(execution_context.session_id)
    print(execution_context.actor_id)
    print(execution_context.parent_execution_id)
    return {"processed": True}

Patterns

Sequential Pipeline

Chain agents where each step feeds the next.

@app.reasoner()
async def pipeline(document: str) -> dict:
    # Step 1: Extract entities
    entities = await app.call("extractor.extract", text=document)

    # Step 2: Classify using extracted entities
    classification = await app.call(
        "classifier.classify",
        entities=entities["entities"],
        text=document,
    )

    # Step 3: Generate report
    report = await app.call(
        "reporter.generate",
        classification=classification,
        entities=entities,
    )

    return report

Error Handling

Cross-agent calls raise exceptions on failure. Handle them explicitly.

@app.reasoner()
async def resilient_call(data: dict) -> dict:
    try:
        result = await app.call("analyzer.process", **data)
        return result
    except Exception as e:
        # Log the failure and return a fallback
        app.note(f"Call to analyzer failed: {e}", ["error"])
        return {"status": "degraded", "error": str(e)}

Workflow DAG Visualization

Every cross-agent call is tracked as an edge in the execution DAG. Query the control plane API to retrieve the full graph:

curl http://localhost:8080/api/ui/v1/workflows/{workflowId}/dag

The response contains every execution node and parent-child relationships, enabling visualization of the complete call tree across agents.

SDK reference

Python -- `await app.call(target, *args, **kwargs)`

Parameter	Type	Description
`target`	`str`	Target in `"node_id.function_name"` format
`*args`	positional	Auto-mapped to target function parameter names in order
`**kwargs`	keyword	Keyword arguments passed directly to the target function

Returns: dict -- always returns a JSON-compatible dictionary.

Positional arguments are mapped to parameter names automatically when calling a local function (same node). For remote calls, positional args are sent as arg_0, arg_1, etc. Keyword arguments are recommended for cross-agent calls.

TypeScript -- `await agent.call(target, input)`

Parameter	Type	Description
`target`	`string`	Target in `"nodeId.functionName"` format
`input`	`any`	Plain object passed as the request body

Returns: any -- the deserialized JSON response from the target.

When the target is on the same agent (nodeId matches), the call executes locally without going through the network, but still emits workflow events for DAG tracking.

Go -- `agent.Call(ctx, target, input)`

Parameter	Type	Description
`ctx`	`context.Context`	Context carrying execution metadata
`target`	`string`	Target in `"nodeId.functionName"` format
`input`	`map[string]any`	Input data for the target function

Returns: (map[string]any, error) -- the result map and any error.

Context headers (X-Run-ID, X-Workflow-ID, X-Session-ID, X-Actor-ID) are extracted from the context and forwarded automatically.