Operator Architecture

Framework-agnostic multi-agent orchestration SDK.

Operator Architecture (OA) manages state, context, sub-agents, and orchestration. It is compatible with any agent runtime — Relay, LangChain, OpenAI Agents, HTTP services, or a plain async function.

Install

uv add operator-architecture
# or
pip install operator-architecture

Runtime dependencies: none (stdlib only).

from operator_architecture import (
    StateMachine,
    Coordinator,
    AgentSpec,
    AgentRequest,
    AgentResult,
    callable_agent,
)

What OA owns vs what you own

Core objects

AgentSpec + AgentRunner

Register any number of sub-agents. Each needs a runner OA will call:

async def research(request: AgentRequest) -> AgentResult:
    # call Relay, LangChain, HTTP, … — OA does not care
    return AgentResult(content=f"Findings for: {request.objective}")

researcher = AgentSpec(
    name="researcher",
    description="Read-only exploration",
    skill="You are a careful researcher. Answer with concrete findings.",
    runner=callable_agent(research),
    model="my-model",  # metadata only
)

Protocol:

class AgentRunner(Protocol):
    async def run(
        self,
        request: AgentRequest,
        *,
        streaming_callback: StreamingCallback = None,
    ) -> AgentResult: ...

AgentRequest carries OpenAI-shaped messages, objective, skill, optional checklist / agent_props.
AgentResult.content is staged as agent_message.

Coordinator

Owns the user-facing skill string and optional runner for sm.run():

coordinator = Coordinator(
    skill="You operate the state machine…",  # default skill provided
    runner=my_coordinator_runner,            # optional
    model="coord-model",
)

StateMachine

sm = StateMachine(coordinator=coordinator, agents=[researcher])

No process-global singleton — hold the instance yourself (sm = StateMachine(...)).

Lifecycle

user → (optional) sm.run / host
     → commission(agent, objective)
         → AgentRunner.run(AgentRequest)
         → stage agent_message  (status=staged)
     → accept_agent_result(agent, index)   # compact result for core thread
        or instruct_agent(agent, index, message)  # continue junior

Direct API (always available)

staged = await sm.commission("researcher", "Find all uses of vLLM")
msg = sm.get_agent_message("researcher", 1)
accepted = sm.accept("researcher", 1)          # or accept_agent_result
# or:
await sm.instruct("researcher", 1, "Also check Dockerfiles")

sm.list_agents()
sm.list_objectives()

Indexed access: sm.agent("researcher")[1].agent_message.

sm.run (optional)

If Coordinator.runner is set, await sm.run(user_text, streaming_callback=...) appends the user message and invokes that runner with:

• metadata["tools"] — orchestration callables
• metadata["tool_schemas"] — OpenAI tools[] schemas

OA does not execute a tool loop. Your runner (Relay, LangChain, …) must invoke those callables when the model requests them.

tools = sm.orchestration_tools()
# list_agents, commission, get_agent_message,
# accept_agent_result, instruct_agent, list_objectives

OpenAI-compatible context

Coordinator and junior threads are lists of chat.completions-style dicts:

{"role": "system"|"user"|"assistant"|"tool", "content": "...", ...}

Helpers: Messages (.system(), .user(), .assistant(), .to_list()).

streaming_callback

Optional observability hook (UI, logs, websockets). Sync or async:

async def on_stream(event: dict) -> None:
    print(event["phase"], event.get("detail", "")[:80])

await sm.commission("researcher", "…", streaming_callback=on_stream)
# or
await sm.run("…", streaming_callback=on_stream)

Phases: start, token, tool_start, tool_result, tool_error, commissioned, done, fail.
Runners may emit events; OA forwards them and also emits lifecycle events around commission.

Writing adapters

OA never imports those libraries. Keep adapters in your host.

Minimal Relay sketch (host code)

class RelayRunner:
    def __init__(self, model, api_key, base_url, tools):
        self.model, self.api_key, self.base_url, self.tools = model, api_key, base_url, tools

    async def run(self, request, *, streaming_callback=None):
        import encode
        messages = encode.Messages()
        for m in request.messages:
            # map dicts into encode.Messages as needed
            ...
        out = await encode.relay_async(
            model=self.model,
            api_key=self.api_key,
            base_url=self.base_url,
            messages=messages,
            tools=self.tools or request.metadata.get("tools"),
        )
        return AgentResult(content=out.content or "", raw=out)

Example

See examples/minimal_callable.py.

License / status

Early SDK (0.2.0). API may evolve; the orchestration contract (commission / stage / accept) is the stable idea.