Client | Agno

client.py

"""
Show how to connect to MCP servers that use the Streamable HTTP transport using our MCPTools class.

Check the README.md file for instructions on how to run these examples.
"""

import asyncio

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

# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------

# 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:
    mcp_tools = MCPTools(
        transport="streamable-http",
        url=server_url,
        refresh_connection=True,  # (Optional) Refresh the MCP connection and tools on each run
    )
    await mcp_tools.connect()
    agent = Agent(
        model=OpenAIResponses(id="gpt-5.5"),
        tools=[mcp_tools],
        markdown=True,
    )
    await agent.aprint_response(input=message, stream=True, markdown=True)
    await mcp_tools.close()

# We can connect to multiple MCP servers at once, even if they use different transports.
# In this example we connect to both our example server (Streamable HTTP transport), and a different server (stdio transport).
async def run_agent_with_multiple_servers(message: str) -> None:
    airbnb_tools = MCPTools(
        command="npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"
    )
    http_tools = MCPTools(
        transport="streamable-http",
        url=server_url,
        refresh_connection=True,  # (Optional) Refresh the MCP connection and tools on each run
    )
    await airbnb_tools.connect()
    await http_tools.connect()
    agent = Agent(
        model=OpenAIResponses(id="gpt-5.5"),
        tools=[airbnb_tools, http_tools],
        markdown=True,
    )
    await agent.aprint_response(input=message, stream=True, markdown=True)
    await airbnb_tools.close()
    await http_tools.close()

# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    asyncio.run(run_agent("Do I have any birthdays this week?"))
    asyncio.run(run_agent("What else is on my calendar this week?"))
    asyncio.run(
        run_agent_with_multiple_servers(
            "Can you check when is my mom's birthday, and if there are any AirBnb listings in SF for two people for that day?",
        )
    )

Run the Example

Set up your virtual environment

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

Install dependencies

uv pip install -U "agno[mcp]" openai

Prepare Node.js

The MCP server runs with npx. Install Node.js, then verify the commands:

node --version
npx --version

Export your OpenAI API key

export OPENAI_API_KEY="your_openai_api_key_here"

Start the companion server

Follow the server setup, including its FastMCP import and port changes. Start that server.py in another terminal using the same activated environment. It listens on http://localhost:8000/mcp. Its calendar responses are demonstration fixtures.

Close connections on failure

Save the source as client.py, then replace both functions with the following. The contexts release each opened connection even if a later connection or model call fails. refresh_connection=True checks connection liveness before a run and reconnects when needed.

async def run_agent(message: str) -> None:
    async with MCPTools(
        transport="streamable-http", url=server_url, refresh_connection=True
    ) as mcp_tools:
        agent = Agent(
            model=OpenAIResponses(id="gpt-5.5"),
            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(command="npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt") as airbnb_tools,
        MCPTools(transport="streamable-http", url=server_url, refresh_connection=True) as http_tools,
    ):
        agent = Agent(
            model=OpenAIResponses(id="gpt-5.5"),
            tools=[airbnb_tools, http_tools],
            markdown=True,
        )
        await agent.aprint_response(input=message, stream=True, markdown=True)

Run the example

Run the adapted client.py:

python client.py

Full source: cookbook/91_tools/mcp/streamable_http_transport/client.py