MCP のトランスポートはメッセージを送受信する方法を定義します。Agno の MCP 統合は SSE トランスポートをサポートしています。このトランスポートは サーバー-to-クライアント・ストリーミングを可能にし、制限されたネットワークで動作する場合 stdio よりも便利なことがわかっています。

Agno : ユーザガイド : コンセプト : ツール – MCP : トランスポート (SSE)

作成 : クラスキャット・セールスインフォメーション
作成日時 : 08/07/2025
バージョン : Agno 1.7.7

* 本記事は docs.agno.com の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています :

• User Guide : Concepts : Tools – MCP : Transports (SSE)

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

Agno ユーザガイド : コンセプト : ツール – MCP : トランスポート (SSE)

Agno の MCP 統合は SSE トランスポート をサポートしています。このトランスポートは サーバー-to-クライアント・ストリーミングを可能にし、制限されたネットワークで動作する場合 stdio よりも役立つことがわかっています。

使用するには、MCP サーバの URL を渡して transport を sse に設定して、MCPTools を初期化します :

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)

server_params 引数を使用して MCP 接続を定義することもできます。以下のように、すべてのリクエストで MCP サーバに送信する headers と timeout 値を指定できます :

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)

    ...

完全な例

単純なローカルサーバをセットアップして SSE トランスポートを使用してそれに接続してみましょう。

1. サーバのセットアップ

   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. クライアントのセットアップ

   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:
       async with MCPTools(transport="sse", url=server_url) as mcp_tools:
           agent = Agent(
               model=OpenAIChat(id="gpt-4o"),
               tools=[mcp_tools],
               markdown=True,
           )
           await agent.aprint_response(message=message, stream=True, markdown=True)
   
   
   # 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:
       async with MultiMCPTools(
           commands=["npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"],
           urls=[server_url],
           urls_transports=["sse"],
       ) as mcp_tools:
           agent = Agent(
               model=OpenAIChat(id="gpt-4o"),
               tools=[mcp_tools],
               markdown=True,
           )
           await agent.aprint_response(message=message, stream=True, markdown=True)
   
   
   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. サーバの実行

   python sse_server.py
   

4. クライアントの実行

   python sse_client.py
   

   

以上