Agent SDK (Python)

pip install 'pipelines-sdk[odyssey]'

pip install 'pipelines-sdk[openai-agents]'
pip install 'pipelines-sdk[anthropic]'
pip install 'pipelines-sdk[langchain]'
pip install 'pipelines-sdk[strands]'

pipelines odyssey init --framework anthropic --dir my-agent

from agents import Agent, Runner, function_tool
from fastapi import FastAPI
from pipelines.odyssey import proxy_call, register_dispatch_route

app = FastAPI()


@function_tool
def get_order(order_id: str) -> dict:
    return proxy_call("get_order", {"order_id": order_id})


@function_tool
def refund_order(order_id: str) -> dict:
    return proxy_call("refund_order", {"order_id": order_id})


def build_agent() -> Agent:
    return Agent(
        name="orders-triage",
        instructions="Triage refund requests; look up orders before deciding.",
        tools=[get_order, refund_order],
        model="gpt-5",
    )


@register_dispatch_route(app, agent_token_env="AGENT_TOKEN")
async def run(envelope):
    result = await Runner.run(build_agent(), envelope.user_instruction)
    return result.final_output

from pipelines.odyssey.adapters.openai_agents import dump_tools_schema_json
print(dump_tools_schema_json(build_agent()))

@dataclass(frozen=True)
class Envelope:
    proxy_url: str
    run_token: str
    user_instruction: str
    task_input: Mapping[str, Any]
    task_id: int | str | None
    run_id: int | str | None
    agent_id: int | str | None
    latest_user_prompt: str | None
    session_id: int | str | None
    turn_id: str | None
    messages: list | None
    scenario_state: Mapping | None
    workspace: Mapping | None
    run_token_jti: str | None
    tools_schema: list[dict] | None
    declared_actor_ids: frozenset[str] | None
    raw: Mapping[str, Any]

# Disable auto-retry for a latency-sensitive call:
result = proxy_call("get_order", {"order_id": "4521"}, auto_retry=False)

from pipelines.odyssey import (
    post_trace_event,
    safe_post_trace_event,
    async_safe_post_trace_event,
)

post_trace_event("system_prompt", {"text": SYSTEM_PROMPT, "model": "gpt-5"})
safe_post_trace_event("thinking", {"text": "Considering refund eligibility..."})
await async_safe_post_trace_event("assistant_message", {"text": "Refund approved."})

from pipelines.odyssey.adapters.openai_agents import (
    dump_tools_schema_json,
    proxy_tool,
    forward_run_result_events,
)

print(dump_tools_schema_json(build_agent()))

get_order = proxy_tool("get_order", description="Look up an order.", is_async=False)

result = await Runner.run(build_agent(), envelope.user_instruction)
forward_run_result_events(result)  # post-run live-trace flush
return result.final_output

from functools import lru_cache

from anthropic import Anthropic
from pipelines.odyssey import register_dispatch_route
from pipelines.odyssey.adapters.anthropic import run_anthropic_loop

TOOLS = [...]  # in `tools_schema` shape — strip registration-only keys
               # (default_execution_mode, passthrough_binding, ledger_*)
               # before passing to the model; messages.create rejects them.


@lru_cache(maxsize=1)
def get_client() -> Anthropic:
    # Lazy: `Anthropic()` raises without ANTHROPIC_API_KEY, so building
    # it at import would crash the server before the ping probe can pass.
    return Anthropic()


@register_dispatch_route(app, agent_token_env="AGENT_TOKEN")
def run(envelope):  # sync def: the SDK offloads it to a worker thread
    return run_anthropic_loop(
        client=get_client(),
        tools=TOOLS,
        user_instruction=envelope.user_instruction,
        model="claude-sonnet-4-5",
        system_prompt=SYSTEM_PROMPT,
        thinking_budget_tokens=1024,
    )

from fastapi import FastAPI
from pydantic import BaseModel, Field
from pipelines.odyssey import register_dispatch_route
from pipelines.odyssey.adapters.langchain import dump_tools_schema, proxy_tool

app = FastAPI()


class GetOrderArgs(BaseModel):
    order_id: str = Field(description="The order id to look up.")


class RefundOrderArgs(BaseModel):
    order_id: str


get_order = proxy_tool("get_order", description="Look up an order.", args_schema=GetOrderArgs)
refund_order = proxy_tool("refund_order", description="Refund an order.", args_schema=RefundOrderArgs)
TOOLS = [get_order, refund_order]


@register_dispatch_route(app, agent_token_env="AGENT_TOKEN")
async def run(envelope):
    from langchain_openai import ChatOpenAI
    from langgraph.prebuilt import create_react_agent

    agent = create_react_agent(ChatOpenAI(model="gpt-5"), TOOLS)
    result = await agent.ainvoke({"messages": [("user", envelope.user_instruction)]})
    return result["messages"][-1].content

print(dump_tools_schema(TOOLS))

from langchain_core.tools import tool
from pipelines.odyssey.adapters.langchain import pipelines_proxy

@pipelines_proxy()
@tool
def get_order(order_id: str) -> dict:
    """Look up an order by id."""
    return db.orders.find(order_id)   # runs locally; proxied at dispatch time

from pipelines.odyssey import pipelines_proxy

from pipelines.odyssey.adapters.strands import (
    dump_tools_schema,
    proxy_tool,
    forward_agent_result_events,
)

get_order = proxy_tool("get_order", description="Look up an order by id.")
print(dump_tools_schema(my_strands_agent))

result = agent(envelope.user_instruction)
forward_agent_result_events(result)  # post-run live-trace flush
return str(result.message) if result.message else str(result)

import os
from pipelines.odyssey import tool_endpoints
from research_agent import tools_network

endpoint = tool_endpoints.create_code(
    name="research-tools",
    functions=[
        tools_network.web_search,
        tools_network.search_wikipedia,
        tools_network.fetch_url,
    ],
    requirements=["httpx>=0.27", "trafilatura"],
    org_id=42,
    api_key=os.environ["PIPELINES_API_KEY"],
)
print(endpoint["id"], endpoint["name"])

payload = tool_endpoints.build_code_endpoint_payload(
    name="research-tools",
    functions=[tools_network.web_search],
    requirements=["httpx>=0.27"],
)

{
  "tools_schema": [
    {
      "name": "search_wikipedia",
      "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
      "default_execution_mode": "passthrough",
      "passthrough_binding": {
        "endpoint_name": "research-tools",
        "tool_name": "search_wikipedia"
      }
    }
  ]
}

endpoint = tool_endpoints.create_code(
    name="research-tools",
    functions=[tools_network.web_search],
    requirements=["httpx>=0.27"],
    env_vars=["TAVILY_API_KEY"],   # ← name of an org credential
    org_id=42,
    api_key=os.environ["PIPELINES_API_KEY"],
)

def web_search(query: str) -> list[dict]:
    key = os.environ["TAVILY_API_KEY"]   # injected by the platform
    ...

curl -X PUT \
  "$PIPELINES_BASE_URL/api/organizations/42/credentials/TAVILY_API_KEY" \
  -H "Authorization: Bearer $PIPELINES_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"value": "tvly-..."}'

import json
import os
from pipelines.odyssey import agents

agent = agents.create_http_agent(
    name="orders-triage",
    endpoint_url="https://your-agent.example.com/dispatch",
    api_key=os.environ["PIPELINES_API_KEY"],
    auth_header_name="Authorization",
    auth_credential_value=f"Bearer {os.environ['AGENT_TOKEN']}",
    tools_schema=json.load(open("tools_schema.json")),
)
print(agent["id"])

from pipelines.odyssey import register_tools_schema_route

register_tools_schema_route(app, TOOLS)  # GET /tools_schema -> {"tools_schema": [...]}

pipelines odyssey sync --from-app http://localhost:8000 --agent-id 42

from pathlib import Path
from pipelines.odyssey import agents

payload = agents.build_code_agent_payload(
    name="research-agent",
    entrypoint="run",
    entrypoint_file="main.py",
    source_dir=Path("./my_agent"),
    requirements=["httpx>=0.27", "trafilatura"],
)

agent = agents.create_code_agent(
    name="research-agent",
    entrypoint="run",
    entrypoint_file="main.py",
    source_dir=Path("./my_agent"),
    requirements=["httpx>=0.27", "trafilatura"],
    api_key="pk_live_...",
)

agents.update_code_agent(
    agent_id=42,
    api_key="pk_live_...",
    source_dir=Path("./my_agent"),
    entrypoint="run",
    entrypoint_file="main.py",
    requirements=["httpx>=0.27"],
)

draft = agents.generate_ledger_schema(api_key="pk_live_...", tools_schema=my_tools)
agents.create_code_agent(..., tools_schema=my_tools, ledger_schema=draft)

from pipelines.odyssey import (
    Envelope,
    EnvelopeError,
    InboundAuthError,
    build_response,
    require_pipelines_auth,
    set_current,
)

def handle_dispatch(headers, body, agent_token):
    try:
        require_pipelines_auth(headers.get("Authorization"), expected_token=agent_token)
    except InboundAuthError:
        return 401, {"detail": "missing or invalid bearer token"}

    try:
        envelope = Envelope.parse(body, headers)
    except EnvelopeError as exc:
        return 400, {"detail": str(exc)}

    with set_current(envelope):
        result = your_agent_runtime(envelope)

    return 200, build_response(result)

@register_dispatch_route(
    app,
    auth_header_name="X-API-Key",
    agent_token_env="AGENT_TOKEN",
)
async def run(envelope):
    ...

from pipelines.odyssey import Envelope, set_current

def test_get_order_calls_proxy(httpx_mock):
    envelope = Envelope(
        proxy_url="http://stub/proxy",
        run_token="t",
        user_instruction="",
        task_input={},
    )
    httpx_mock.add_response(
        url="http://stub/proxy/tools/get_order",
        json={"tool_name": "get_order", "response": {"id": "1"}, "source": "odyssey"},
    )
    with set_current(envelope):
        result = get_order("1")
    assert result == {"id": "1"}