Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs-v1.agno.com/llms.txt

Use this file to discover all available pages before exploring further.

Agno’s MCP integration supports the SSE transport. This transport enables server-to-client streaming, and can prove more useful than stdio when working with restricted networks. To use it, initialize the MCPTools passing the URL of the MCP server and setting the transport to sse: 
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools

server_url = "http://localhost:8000/sse"

async with MCPTools(url=server_url, transport="sse") as mcp_tools:
    agent = Agent(model=OpenAIChat(id="gpt-4o"), tools=[mcp_tools])
    await agent.aprint_response("What is the license for this project?", stream=True)
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: 
from agno.tools.mcp import MCPTools, SSEClientParams

server_params = SSEClientParams(
    url=...,
    headers=...,
    timeout=...,
    sse_read_timeout=...,
)

async def run_mcp_agent():

    # Initialize the MCP tools with the server parameters
    mcp_tools = MCPTools(server_params=server_params)

    ...

​
Complete example

Let’s set up a simple local server and connect to it using the SSE transport:
1

Setup the server

sse_server.py
from mcp.server.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="sse")
2

Setup the client

sse_client.py
import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools, MultiMCPTools

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


async def run_agent(message: str) -> None:
    mcp_tools = MCPTools(transport="sse", url=server_url)
    await mcp_tools.connect()

    agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        tools=[mcp_tools],
        markdown=True,
    )
    await agent.aprint_response(message=message, stream=True, markdown=True)

    await mcp_tools.close()


# Using MultiMCPTools, 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 (SSE transport), and a different server (stdio transport).
async def run_agent_with_multimcp(message: str) -> None:

    # Initialize the MultiMCPTools instance
    multi_mcp_tools =MultiMCPTools(
        commands=["npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"],
        urls=[server_url],
    )

    # Connect to the MCP servers
    await multi_mcp_tools.connect()

    # Initialize the Agent
    agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        tools=[mcp_tools],
        markdown=True,
    )

    # Run the agent
    await agent.aprint_response(message=message, stream=True, markdown=True)

    # Close the MCP connections
    await multi_mcp_tools.close()


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

Run the server

python sse_server.py
4

Run the client

python sse_client.py