Quickstart: OpenAI Agents SDK

The OpenAI Agents SDK supports MCP servers as tool sources. Pipeworx plugs in like any other MCP.

Install

pip install openai-agents

Wire it up

Pipeworx is a streamable-HTTP MCP server, not classic SSE — MCPServerSse opens a GET connection first, which the gateway rejects (405; it only accepts POST at /mcp). Use MCPServerStreamableHttp instead:

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async with MCPServerStreamableHttp(
    name="pipeworx",
    params={"url": "https://gateway.pipeworx.io/mcp"},
) as pipeworx:
    agent = Agent(
        name="ResearchAgent",
        instructions="You are a research assistant. Use Pipeworx for live data.",
        mcp_servers=[pipeworx],
    )
    result = await Runner.run(agent, "What is the current 30-year mortgage rate?")
    print(result.final_output)

The agent automatically calls pipeworx’s tools when relevant. The gateway’s serverInstructions brief the model on ask_pipeworx, discover_tools, etc.

Scope by task

For tighter context budgets:

pipeworx_housing = MCPServerStreamableHttp(
    name="pipeworx-housing",
    params={"url": "https://gateway.pipeworx.io/mcp?vertical=housing"},
)

Only housing-relevant tools surface in tools/list — far fewer than the full catalog. See context tax.

Multiple gateways

You can mount Pipeworx multiple times with different scopes. Each server needs its own connect() (or async with) before the agent runs:

pw_finance  = MCPServerStreamableHttp(name="pw-finance",  params={"url": "https://gateway.pipeworx.io/mcp?vertical=fintech"})
pw_pharma   = MCPServerStreamableHttp(name="pw-pharma",   params={"url": "https://gateway.pipeworx.io/mcp?vertical=pharma"})
pw_research = MCPServerStreamableHttp(name="pw-research", params={"url": "https://gateway.pipeworx.io/mcp?task=academic+research"})

for server in (pw_finance, pw_pharma, pw_research):
    await server.connect()

agent = Agent(
    ...,
    mcp_servers=[pw_finance, pw_pharma, pw_research],
)

# later: await server.cleanup() for each, or manage them with contextlib.AsyncExitStack

The agent SDK namespaces tools by server, so there’s no name collision even with overlapping pack content.

Authentication

Pass headers via params:

MCPServerStreamableHttp(
    name="pipeworx",
    params={
        "url": "https://gateway.pipeworx.io/mcp",
        "headers": {"Authorization": f"Bearer {pipeworx_token}"}
    },
)

Pipeworx-aware system prompt

The gateway’s serverInstructions are excellent agent guidance (“use ask_pipeworx for plain English routing”, “prefer compound _intel tools”, etc.). The Agents SDK reads them automatically on connect, but if you want to reinforce them in the agent’s primary instructions:

async with MCPServerStreamableHttp(
    name="pipeworx",
    params={"url": "https://gateway.pipeworx.io/mcp"},
) as pipeworx:
    # connect() already ran initialize() internally and cached the result —
    # don't call session.initialize() again, just reuse the session.
    instructions = pipeworx.server_initialize_result.instructions
    agent = Agent(
        name="ResearchAgent",
        instructions=f"You are a research assistant.\n\n{instructions}",
        mcp_servers=[pipeworx],
    )

Without this, the SDK still passes instructions to the model — but as a separate field that some model versions weight less than the primary instructions.

Reading response metadata

Every Pipeworx tool response includes _meta with cost, freshness, retry hints, and (on errors) examples and alternatives. There’s no @agent.on_tool_call_complete decorator in the SDK — hooks are a class you attach, overriding on_tool_end:

import json
from agents import AgentHooks

class MetaLoggingHooks(AgentHooks):
    async def on_tool_end(self, context, agent, tool, result):
        # `result` is the tool's return value — a JSON string for most MCP
        # tools, occasionally a dict; handle both defensively.
        try:
            data = json.loads(result) if isinstance(result, str) else result
        except (TypeError, ValueError):
            return
        meta = data.get("_meta") if isinstance(data, dict) else None
        if not meta:
            return
        if "cache" in meta:
            print(f"  freshness: {meta['cache'].get('fresh_until')}")
        if "feedback_hint" in meta:
            print(f"  feedback: {meta['feedback_hint']}")

agent = Agent(
    name="ResearchAgent",
    instructions="...",
    mcp_servers=[pipeworx],
    hooks=MetaLoggingHooks(),
)

Useful for production agents that need to surface freshness to humans or auto-file feedback on errors.

Memory across runs

Pipeworx exposes remember / recall / forget as tools. Agents authenticated as the same account share state across runs:

# Run 1
await Runner.run(agent, "Remember the focus company for this research is AAPL.")

# Later, separate run, same account
result = await Runner.run(agent, "Pick up where I left off — the focus company.")
# Agent calls recall({key: "..."}) and continues

See memory.

Caveats

  • MCPServerStreamableHttp calls list_tools() on the server every turn by default (cache_tools_list=False). Pass cache_tools_list=True in the constructor once you know the tool set is stable, to skip the round-trip.
  • _meta.cache.fresh_until in tools/call responses — your logic can decide whether to re-call vs. trust prior data. See caching and freshness.
  • The SDK does not auto-retry a failed tool call — a tool error is just appended to the conversation and it’s the model’s call whether to try again. Pipeworx returns _meta.examples and _meta.alternatives on errors specifically so the model has enough in context to self-correct instead of repeating the same failing call. See error recovery.

Last reviewed August 10, 2026