orchagent SDK

Official Python SDK for building orchagent agents with agent-to-agent call support.

Installation

pip install orchagent-sdk

Note: The pip package is orchagent-sdk but the Python import is from orchagent import .... This is a common Python convention (e.g. Pillow → import PIL).

Quick Start

# pip install orchagent-sdk  (package name)
from orchagent import AgentClient  # module name

# Create a client (uses ORCHAGENT_SERVICE_KEY env var)
client = AgentClient()

# Call another agent
async def my_orchestrator(input_data):
    result = await client.call("joe/leak-finder@v1", {"url": input_data["repo"]})
    return result

Features

• Agent-to-agent calls - Call other agents from within your orchestrator agent
• Service key authentication - Automatic auth via ORCHAGENT_SERVICE_KEY env var
• Call chain tracking - Prevents circular dependencies (A calls B calls A)
• Deadline propagation - Timeout context flows through the call chain
• Local execution mode - Test orchestrators locally with orch run --with-deps

Usage

Basic Usage

from orchagent import AgentClient

client = AgentClient()
result = await client.call("org/agent@v1", {"input": "data"})
# result is the raw agent output — the gateway envelope is auto-unwrapped

With FastAPI

Extract call context from incoming requests:

from fastapi import Request
from orchagent import AgentClient

@app.post("/analyze")
async def analyze(request: Request, input: AnalyzeInput):
    # Automatically extracts call chain, deadline, etc. from request headers
    client = AgentClient.from_request(request)
    secrets = await client.call("joe/leak-finder@v1", {"url": input.repo_url})
    return {"secrets": secrets}

Error Handling

from orchagent import (
    AgentClient,
    DependencyCallError,
    CallChainCycleError,
    TimeoutExceededError,
)

try:
    result = await client.call("org/agent@v1", data)
except CallChainCycleError:
    # Would create a circular dependency
    pass
except TimeoutExceededError:
    # Deadline passed
    pass
except DependencyCallError as e:
    # Agent returned an error
    print(f"Status: {e.status_code}, Body: {e.response_body}")

Local Execution Mode

When running with orch run --with-deps, the SDK automatically detects local mode and spawns sub-agents as local subprocesses instead of making HTTP calls.

# Agents are executed locally as subprocesses
orch run orchagent/security-review --with-deps --input '{"path": "."}'

This is controlled by the ORCHAGENT_LOCAL_EXECUTION=true environment variable, which the CLI sets automatically.

Environment Variables

API Reference

AgentClient

class AgentClient:
    def __init__(
        self,
        service_key: str | None = None,      # From ORCHAGENT_SERVICE_KEY
        gateway_url: str | None = None,       # From ORCHAGENT_GATEWAY_URL
        call_chain: list[str] | None = None,  # Current call chain
        deadline_ms: int | None = None,       # Deadline timestamp
        max_hops: int | None = None,          # Max remaining hops
    ): ...

    @classmethod
    def from_request(cls, request, service_key=None) -> "AgentClient": ...

    async def call(
        self,
        agent_ref: str,           # "org/agent@version"
        input_data: dict,         # Input payload
        endpoint: str | None,     # Optional endpoint override
        timeout: float | None,    # Optional timeout in seconds
    ) -> Any: ...
    # Returns the raw agent output, not the gateway envelope.
    # The gateway wraps responses in {"data": ..., "metadata": ...} —
    # the SDK auto-unwraps this so you get the agent's output directly.

Exceptions

• AgentClientError - Base exception
• DependencyCallError - Agent call failed (has status_code, response_body)
• CallChainCycleError - Would create circular dependency
• TimeoutExceededError - Deadline passed
• LocalExecutionError - Subprocess execution failed (has exit_code, stderr)

License

MIT