HTTP ストリーミング (Streamable HTTP) トランスポートが、プロトコルバージョン 2024-11-05 から HTTP+SSE トランスポートに置き換わります。このトランスポートは MCP サーバが複数のクライアント接続を処理することを可能にし、サーバ-to-クライアント・ストリーミングには SSE を使用することもできます。

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

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

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

• User Guide : Concepts : Tools – MCP : Transports (Streamable HTTP)

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

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

新しい HTTP ストリーミング (Streamable HTTP) トランスポートが、プロトコルバージョン 2024-11-05 から HTTP+SSE トランスポートに置き換わります。

このトランスポートは MCP サーバが複数のクライアント接続を処理することを可能にし、サーバ-to-クライアント・ストリーミングには SSE を使用することもできます。

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

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

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

async def run_mcp_agent():

    # Initialize the MCP tools
    mcp_tools = MCPTools(url=server_url, transport="streamable-http")

    # Connect to the MCP server
    await mcp_tools.connect()

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

    # Run the agent
    await agent.aprint_response("What is the license for this project?", stream=True)

    # Close the MCP connection
    await mcp_tools.close()

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

from agno.tools.mcp import MCPTools, StreamableHTTPClientParams

server_params = StreamableHTTPClientParams(
    url=...,
    headers=...,
    timeout=...,
    sse_read_timeout=...,
    terminate_on_close=...,

)

async def run_mcp_agent():

    # Initialize the MCP tools
    mcp_tools = MCPTools(server_params=server_params)

    # Connect to the MCP server
    await mcp_tools.connect()

    ...

完全な例

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

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

   streamable_http_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="streamable-http")
   

2. クライアントのセットアップ

   streamable_http_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/mcp"
   
   
   async def run_agent(message: str) -> None:
       async with MCPTools(transport="streamable-http", 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 (Streamable HTTP 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=["streamable-http"],
       ) 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?",
           )
       )
   

   

以上