Streamable HTTP MCP Server

Streamable HTTP MCP Server Skill

This skill helps create and configure Streamable HTTP Model Context Protocol (MCP) server connections for OpenAI Agents SDK.

Purpose

• Create MCPServerStreamableHttp configurations
• Configure HTTP connection parameters and authentication
• Set up caching and retry mechanisms
• Connect to HTTP-based MCP servers with direct connection management

MCPServerStreamableHttp Constructor Parameters

• params (MCPServerStreamableHttpParams): Connection parameters for the server
  • url (str): The URL of the server
  • headers (dict[str, str], optional): The headers to send to the server
  • timeout (timedelta | float, optional): The timeout for the HTTP request (default: 5 seconds)
  • sse_read_timeout (timedelta | float, optional): The timeout for the SSE connection (default: 5 minutes)
  • terminate_on_close (bool, optional): Whether to terminate on close
  • httpx_client_factory (HttpClientFactory, optional): Custom HTTP client factory for configuring httpx.AsyncClient behavior
• cache_tools_list (bool): Whether to cache the list of available tools (default: False)
• name (string | None): A readable name for the server (default: None, auto-generated from URL)
• client_session_timeout_seconds (float | None): Read timeout for the MCP ClientSession (default: 5)
• tool_filter (ToolFilter): The tool filter to use for filtering tools (default: None)
• use_structured_content (bool): Whether to use tool_result.structured_content when calling an MCP tool (default: False)
• max_retry_attempts (int): Number of times to retry failed list_tools/call_tool calls (default: 0)
• retry_backoff_seconds_base (float): The base delay, in seconds, for exponential backoff between retries (default: 1.0)
• message_handler (MessageHandlerFnT | None): Optional handler invoked for session messages (default: None)

Usage Context

Use this skill when:

• Managing HTTP connections yourself
• Running servers locally or remotely with direct connection management
• Needing to keep latency low with your own infrastructure
• Wanting to run the server inside your own infrastructure

Basic Example

import asyncio
import os

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

async def main() -> None:
    token = os.environ["MCP_SERVER_TOKEN"]
    async with MCPServerStreamableHttp(
        name="Streamable HTTP Python Server",
        params={
            "url": "http://localhost:8000/mcp",
            "headers": {"Authorization": f"Bearer {token}"},
            "timeout": 10,
        },
        cache_tools_list=True,
        max_retry_attempts=3,
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions="Use the MCP tools to answer the questions.",
            mcp_servers=[server],
            model_settings=ModelSettings(tool_choice="required"),
        )

        result = await Runner.run(agent, "Add 7 and 22.")
        print(result.final_output)

asyncio.run(main())