Streamable HTTP Transport

Connect MCPTools to an MCP server over Streamable HTTP using the url and transport parameters.

The new Streamable HTTP transport replaces the HTTP+SSE transport from protocol version 2024-11-05.

This transport enables the MCP server to handle multiple client connections, and can also use SSE for server-to-client streaming.

To use it, initialize the MCPTools passing the URL of the MCP server and setting the transport to streamable-http:

Prerequisites

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install Node.js for the npx server used in the complete example.

uv pip install -U "agno[mcp]" openai
export OPENAI_API_KEY="your_openai_api_key_here"
node --version
npx --version
import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.tools.mcp import MCPTools

async def main():
    async with MCPTools(
        url="https://docs.agno.com/mcp",
        transport="streamable-http",
    ) as mcp_tools:
        agent = Agent(model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools])
        await agent.aprint_response(
            "What can you tell me about MCP support in Agno?",
            stream=True,
        )

asyncio.run(main())

You can also use the server_params argument to define the MCP connection. This way you can specify the headers to send to the MCP server with every request, and the timeout values:

import asyncio

from agno.tools.mcp import MCPTools, StreamableHTTPClientParams

async def main():
    server_params = StreamableHTTPClientParams(
        url="https://docs.agno.com/mcp",
        headers={"Authorization": "Bearer your-token"},
        timeout=30,
        sse_read_timeout=300,
        terminate_on_close=True,
    )

    async with MCPTools(
        server_params=server_params,
        transport="streamable-http",
    ) as mcp_tools:
        print([tool.name for tool in mcp_tools.functions.values()])

asyncio.run(main())

Complete example

Set up a simple local server and connect to it using the Streamable HTTP transport:

Setup the server

from fastmcp import FastMCP

mcp = FastMCP("calendar_assistant")

@mcp.tool()
def get_events(day: str) -> str:
    return f"There are no events scheduled for {day}."

@mcp.tool()
def get_birthdays_this_week() -> str:
    return "It is your mom's birthday tomorrow"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Setup the client

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.tools.mcp import MCPTools

# This is the URL of the MCP server we want to use.
server_url = "http://localhost:8000/mcp"

async def run_agent(message: str) -> None:
    async with MCPTools(
        transport="streamable-http",
        url=server_url,
    ) as mcp_tools:
        agent = Agent(
            model=OpenAIResponses(id="gpt-5.2"),
            tools=[mcp_tools],
            markdown=True,
        )
        await agent.aprint_response(input=message, stream=True, markdown=True)

async def run_agent_with_multiple_servers(message: str) -> None:
    async with (
        MCPTools(transport="streamable-http", url=server_url) as calendar_tools,
        MCPTools(
            command="npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"
        ) as airbnb_tools,
    ):
        agent = Agent(
            model=OpenAIResponses(id="gpt-5.2"),
            tools=[calendar_tools, airbnb_tools],
            markdown=True,
        )
        await agent.aprint_response(input=message, stream=True, markdown=True)

if __name__ == "__main__":
    asyncio.run(run_agent("Do I have any birthdays this week?"))
    asyncio.run(
        run_agent_with_multiple_servers(
            "Check when my mom's birthday is and find Airbnb listings in San Francisco for two people that day."
        )
    )

Run the server in one terminal

python streamable_http_server.py

Run the client in a second terminal

source .venv/bin/activate
export OPENAI_API_KEY="your_openai_api_key_here"
python streamable_http_client.py