Courier OS SDK

Python SDK for building agents against any OpenAI-compatible chat API, with an automatic tool loop, durable sessions, and mid-loop intercept. Licensed builders can run the same Model / Agent / Operator stack on Apple Silicon via optional local MLX inference.

from courier_os import Agent, Model, Operator, tools

@tools.tool
def get_weather(city: str) -> dict:
    """Get weather for a city."""
    return {"city": city, "temp_f": 72}

model = Model("gpt-4o", api_key="…", base_url="https://api.openai.com/v1")
out = Operator(Agent(model, tools=[get_weather])).relay(
    [{"role": "user", "content": "Weather in Denver?"}]
)
print(out.content)

The model called get_weather, the SDK dispatched it, fed the result back, and returned the final answer.

Install: pip install courier-os or uv add courier-os
Import: import courier_os
Python: 3.11+
License: Apache-2.0

What this is

Primitive	Role
Model	Named model. mode="api" (credentials) or mode="local" (optional binary). can_load probes headroom or credentials.
tools	Python callables or OpenAI tool dicts → schemas + dispatch index.
Agent	Model + tools. One model turn; you handle tool_calls if you want a manual loop. Optional session=.
Operator	Auto tool loop: Operator(agent).relay(messages). HTTP is Model(..., api_key, base_url).
Session	Append-only event log and live tool registry. Persist with model_dump(), resume with model_validate().

Works with OpenAI, Courier, vLLM, LM Studio, Ollama, Together, Groq, and any other OpenAI-compatible /v1/chat/completions endpoint.

Install

pip install courier-os
# or
uv add courier-os

Credentials (kwargs or environment; .env is loaded automatically):

COURIER_OS_API_KEY=sk-…
COURIER_OS_BASE_URL=https://api.openai.com/v1

OPENAI_API_KEY / OPENAI_BASE_URL are used if COURIER_OS_* are unset.

Mode	How	Extra packages
API	Model("gpt-4o", api_key="…", base_url="…")	None
Local	Model("mlx-community/…", mode="local")	courier-os-binary (Apple Silicon / MLX; not on public PyPI yet)

The extra courier-os[local] declares a dependency on courier-os-binary. Until that package is published, install the binary from the courier-os-sdk monorepo.

60-second tour

from courier_os import Agent, Model, Operator, Session, tools

model = Model("gpt-4o", api_key="…", base_url="https://api.openai.com/v1")

# 1. Plain chat
reply = model.chat([{"role": "user", "content": "hi"}])
print(reply.content)

# 2. Agent — one turn (you run tools if the model asked)
@tools.tool
def lookup(q: str) -> dict:
    """Look something up."""
    return {"q": q, "answer": "42"}

agent = Agent(model, tools=[lookup])
messages = [{"role": "user", "content": "What's the meaning of life?"}]
resp = agent.chat(messages)
if resp.tool_calls:
    messages.append({"role": "assistant", "content": resp.content, "tool_calls": resp.tool_calls})
    messages.extend(agent.call_tools(resp.tool_calls))
    resp = agent.chat(messages)
print(resp.content)

# 3. Operator — the loop runs until the model is done
out = Operator(agent).relay([{"role": "user", "content": "What's the meaning of life?"}])
print(out.content)

# 4. Session — durable log you persist anywhere
session = Session.open()
agent = Agent(model, tools=[lookup], session=session)
Operator(agent).relay([{"role": "user", "content": "remember: my name is Alex"}])
db_row = session.model_dump()          # file, Postgres, Redis, S3, …

session = Session.model_validate(db_row)
agent = Agent(model, session=session)
Operator(agent).relay([{"role": "user", "content": "what's my name?"}])

The rest of this page reuses model (and get_weather where a tool is needed). Point Model(...) at your endpoint.

Tools

Pass Python functions. Type hints and the short docstring become the OpenAI tool schema. Raw OpenAI dicts work too.

from courier_os import tools

@tools.tool
def get_weather(city: str, units: str = "f") -> dict:
    """Get weather for a city.

    Args:
        city: City name.
        units: Temperature units, f or c.
    """
    return {"city": city, "temp": 72, "units": units}

schema_only = {
    "type": "function",
    "function": {
        "name": "search",
        "description": "Search an index.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
}

agent = Agent(model, tools=[get_weather, schema_only])

@tools.tool is optional (it marks the callable). tools.build([get_weather]) returns Tools(schemas=…, index=…).

Operator

Operator(agent).relay(messages) executes tools, appends results, and calls the model again until there are no more tool calls (or you stop it).

from courier_os import Agent, Operator, tools

@tools.tool
def get_weather(city: str) -> dict:
    """Get weather for a city."""
    return {"city": city, "temp_f": 72}

out = Operator(Agent(model, tools=[get_weather])).relay(
    [{"role": "user", "content": "Weather in Denver?"}],
    temperature=0.2,
    max_tokens=1024,
)
print(out.content)
print(out.tool_calls)   # every call the loop made
print(out.iterations)
print(out.usage)

Per-turn provider fields that are not first-class Operator kwargs go in extra_body. They are merged onto the chat-completions JSON (not nested under "extra_body"):

out = Operator(agent).relay(
    [{"role": "user", "content": "Think step by step."}],
    extra_body={"chat_template_kwargs": {"enable_thinking": True}},
)

Intercept

on_intercept runs after each tool-call turn, before the next model request. Use it to stop the loop, rewrite history, or register tools on a Session.

from courier_os import Agent, Operator, tools

@tools.tool
def submit_final(answer: str) -> dict:
    """Submit the final answer and stop."""
    return {"answer": answer}

def watcher(event):
    if any(tc.name == "submit_final" for tc in event.tool_calls):
        event.stop()

def redact(event):
    event.edit_last_tool_result(lambda c: c[:1000])

def trim(event):
    event.compact(lambda msgs: msgs[-20:])

def inject(event):
    event.append({"role": "system", "content": "Be concise."})

agent = Agent(model, tools=[get_weather, submit_final])
Operator(agent, on_intercept=watcher).relay(
    [{"role": "user", "content": "Weather in Denver, then submit."}]
)

InterceptEvent fields: iteration, assistant_turn, tool_calls, raw_response, will_continue, messages, session, mutated, stopped.

Mutators: stop(), append, insert, replace, compact, edit_last_tool_result, register_tool (requires a Session).

Loop limit

from courier_os import MaxToolIterationsError

try:
    Operator(agent, max_tool_iterations=8).relay(
        [{"role": "user", "content": "Keep calling tools until you finish."}]
    )
except MaxToolIterationsError as exc:
    print(exc.partial.content)   # conversation so far

Session

A Session is an append-only event log outside the context window. The Operator hydrates Messages from events, then writes new events as the loop runs. Storage is yours: model_dump() / model_validate().

from courier_os import Agent, Operator, Session

session = Session.open(id="user-42", metadata={"user": "alex"}, tools=[get_weather])
agent = Agent(model, session=session)
Operator(agent).relay([{"role": "user", "content": "Weather in Denver?"}])

# Types in the log, in order
[e.type for e in session.events]
# user.message, assistant.message, tool.call, tool.result, iteration.end, …

session.events_by_type("tool.call")
session.events_since(n)          # cursor: id > n
session.to_messages()            # context window for the model

# Custom analytics never enter the prompt
session.emit("billing.charge", {"cents": 42})

Attach the session on the Agent. Operator.relay() uses agent.session and the live session.tools list. Operator.relay(..., session=other) overrides. Explicit tools= still wins.

Resume

import json

raw = json.loads(json.dumps(session.model_dump(), default=str))
session = Session.model_validate(raw)          # tools re-imported from tool.registered
# or Session.resume(raw, tools=[my_closure])  # fill lambdas / __main__ functions

agent = Agent(model, session=session)
Operator(agent).relay([{"role": "user", "content": "and Boston?"}])

session.tools is excluded from JSON. Module-level callables restore via "module:qualname". Lambdas, __main__ functions, and moved modules land in session.unresolved_tools — pass them to rebind_tools or Session.resume(..., tools=…). Do not validate dumps from untrusted sources (importlib is used on resume).

AsyncSession adds aemit, aregister_tool, aregister_tools, arebind_tools.

Mid-loop tool unlock

Register new tools from an intercept. They appear on the next model turn.

from courier_os import Agent, Operator, Session, tools

@tools.tool
def list_tools() -> list[dict]:
    """Discover available tools."""
    return [{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get weather for a city.",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    }]

@tools.tool
def get_weather(city: str) -> dict:
    """Get weather for a city."""
    return {"city": city, "temp_f": 72}

IMPLS = {"get_weather": get_weather}

def unlock(event):
    for tc in event.tool_calls:
        if tc.name == "list_tools":
            for spec in tc.result or []:
                name = spec["function"]["name"]
                if name in IMPLS:
                    event.register_tool(IMPLS[name])

session = Session.open(tools=[list_tools])
agent = Agent(model, session=session)
Operator(agent, on_intercept=unlock).relay(
    [{"role": "user", "content": "List tools, then weather in Denver."}]
)

Same-name register_tool is a no-op (False). Intercept mutations also emit context.modify on the session.

Messages

Messages is a mutable OpenAI-shaped conversation. Pass the same object to Operator.relay and the SDK replaces it with the full history after a non-streaming run. Plain lists are not mutated.

from courier_os import Messages, Operator

m = Messages().system("Be brief.").user("name three colors")
Operator(agent).relay(m)
m.user("now three more")
Operator(agent).relay(m)

Helpers: .system(), .user(), .assistant(), .tool(content, tool_call_id=…), .add(), .to_list(), .to_pydantic(), Messages.from_pydantic(), Messages.from_events().

Streaming

Operator.relay(..., stream=True) yields loop events (content.delta, tool_calls.delta, finish) while the tool loop is intact:

for event in Operator(agent).relay(
    [{"role": "user", "content": "Weather in Denver?"}],
    stream=True,
):
    if event.type == "content.delta":
        print(event.data, end="", flush=True)

Model-level token stream (no tool loop):

for delta in model.chat_stream([{"role": "user", "content": "hi"}]):
    if delta.delta:
        print(delta.delta, end="", flush=True)
    if delta.done:
        print(delta.content)

response_format cannot be combined with stream=True.

Async

relay_async is a normal method that returns an awaitable, async-iterable handle:

out = await Operator(agent).relay_async([{"role": "user", "content": "Weather in Denver?"}])

async for event in Operator(agent).relay_async(
    [{"role": "user", "content": "Weather in Denver?"}],
    stream=True,
):
    ...

reply = await model.chat_async([{"role": "user", "content": "hi"}])
async for delta in model.chat_stream_async([{"role": "user", "content": "hi"}]):
    ...

Async tool callables and async intercept callbacks work on the async path. Use relay_async when the interceptor is async.

Multimodal

OpenAI-shaped image and audio content parts work on API and local omni models:

from courier_os import AudioContent, ImageContent

messages = [{
    "role": "user",
    "content": [
        {"type": "text", "text": "What is in this image?"},
        {"type": "image_url", "image_url": {"url": "data:image/png;base64,…"}},
    ],
}]
Operator(agent).relay(messages)

# Helpers
ImageContent.from_path("shot.png")
ImageContent.from_url("https://example.com/shot.png")
AudioContent.from_path("clip.wav")

Structured output

Pass a Pydantic model (or a JSON-schema dict) as response_format. The parsed instance is out.parsed when JSON validates.

from pydantic import BaseModel

class Weather(BaseModel):
    city: str
    temp_f: float

out = Operator(agent).relay(
    [{"role": "user", "content": "Weather in Denver as JSON."}],
    response_format=Weather,
)
print(out.parsed)

Custom tool execution

By default tools run in-process (LocalToolExecutor). Pass tool_executor= to dispatch remotely, in a sandbox, over MCP, or to a sub-agent. Implementations must return ExecutionResult (serialize errors; do not raise for tool failures).

import json
from courier_os import ExecutionResult, Operator

class RemoteExecutor:
    def execute(self, name: str, input: dict) -> ExecutionResult:
        payload = call_my_tools_service(name, input)
        return ExecutionResult(
            result=payload,
            result_serialized=json.dumps(payload),
            error=None,
            duration_ms=0.0,
        )

    async def execute_async(self, name: str, input: dict) -> ExecutionResult:
        return self.execute(name, input)

Operator(agent, tool_executor=RemoteExecutor()).relay(
    [{"role": "user", "content": "Weather in Denver?"}]
)

Local models and can_load

model = Model("mlx-community/Qwen3.5-2B-MLX-8bit", mode="local", load=False)
if not model.can_load:
    model = Model("gpt-4o", api_key="…", base_url="https://api.openai.com/v1")

print(model.mode, model.name)
agent = Agent(model, tools=[get_weather])

mode is explicit mode= → else credentials/client → else binary available → else error. Local inference requires Apple Silicon, courier-os-binary, and weights under ~/.courier/models.

Errors

All SDK errors subclass CourierOSError (CourierError is an alias).

Exception	When
AuthError	Missing or rejected credentials (401/403)
InvalidRequestError	4xx from the server or bad SDK arguments
InvalidToolCallError	Tool arguments were not valid
RateLimitError	429
ServerError	5xx
TransportError	Network failed after retries
ModelLoadError	Local model failed to load
InferenceError	Local or API inference failed
MaxToolIterationsError	Tool loop hit max_tool_iterations (see .partial)

from courier_os import AuthError, CourierOSError

try:
    Operator(agent).relay([{"role": "user", "content": "hi"}])
except AuthError:
    ...
except CourierOSError as exc:
    print(exc.status, exc.message)

HTTP client

Reuse a client across calls:

from courier_os import Client, Model

client = Client(api_key="…", base_url="https://api.openai.com/v1", timeout=60.0)
model = Model("gpt-4o", client=client)

AsyncClient for the async path. timeout, max_retries apply to HTTP.

API index

Symbol
Model	Dual-mode chat / stream
Agent	Single-turn tools
Operator	Auto tool loop (relay / relay_async)
Session / AsyncSession	Event log + live tools
Event / EventType	Log envelope and standard type strings
InterceptEvent	Mid-loop observe / mutate / stop
Messages / Message / Conversation	Context window
TextContent / ImageContent / AudioContent / ImageURL / InputAudio	Multimodal parts
ToolCall / ToolCallFunction	Typed tool-call shapes
tools / Tools / tool / build_tools	Schema + index
ToolExecutor / LocalToolExecutor / ExecutionResult	Dispatch seam
Client / AsyncClient	HTTP
ChatResponse / StreamDelta / StreamEvent / Usage	Model and stream results
RelayResponse / AssistantTurn / ToolCallRecord	Operator.relay results
CourierOSError and subclasses	Typed errors

License

Apache License 2.0.

Further reading

• Sessions
• Getting started
• Architecture
• Repository