orchagent-sdk 0.1.1

SDK for building orchagent agents - enables agent-to-agent calls with local execution support

orchagent SDK

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

Installation

pip install orchagent-sdk

Quick Start

from orchagent import AgentClient

# 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"})

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

Variable	Description
ORCHAGENT_SERVICE_KEY	Service key for API authentication
ORCHAGENT_GATEWAY_URL	Gateway URL (default: https://api.orchagent.io)
ORCHAGENT_LOCAL_EXECUTION	Enable local subprocess execution
ORCHAGENT_AGENTS_DIR	Path to local agents (default: ~/.orchagent/agents)

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: ...

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