Streamable HTTP Transport

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:

1from kern.agent import Agent
2from kern.models.openai import OpenAIResponses
3from kern.tools.mcp import MCPTools
4
5# Initialize and connect to the Streamable HTTP MCP server
6mcp_tools = MCPTools(url="https://kern.ndx.rocks/mcp", transport="streamable-http")
7await mcp_tools.connect()
8
9try:
10    agent = Agent(model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools])
11    await agent.aprint_response("What can you tell me about MCP support in Kern?", stream=True)
12finally:
13    # Always close the connection when done
14    await mcp_tools.close()

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:

1from kern.tools.mcp import MCPTools, StreamableHTTPClientParams
2
3server_params = StreamableHTTPClientParams(
4    url=...,
5    headers=...,
6    timeout=...,
7    sse_read_timeout=...,
8    terminate_on_close=...,
9)
10
11# Initialize and connect using server parameters
12mcp_tools = MCPTools(server_params=server_params, transport="streamable-http")
13await mcp_tools.connect()
14
15try:
16    # Use mcp_tools with your agent
17    pass
18finally:
19    await mcp_tools.close()

Complete example

Let's set up a simple local server and connect to it using the Streamable HTTP transport:

Setup the server

1from mcp.server.fastmcp import FastMCP
2
3mcp = FastMCP("calendar_assistant")
4
5
6@mcp.tool()
7def get_events(day: str) -> str:
8    return f"There are no events scheduled for {day}."
9
10
11@mcp.tool()
12def get_birthdays_this_week() -> str:
13    return "It is your mom's birthday tomorrow"
14
15
16if __name__ == "__main__":
17    mcp.run(transport="streamable-http")

Setup the client

1import asyncio
2
3from kern.agent import Agent
4from kern.models.openai import OpenAIResponses
5from kern.tools.mcp import MCPTools, MultiMCPTools
6
7# This is the URL of the MCP server we want to use.
8server_url = "http://localhost:8000/mcp"
9
10
11async def run_agent(message: str) -> None:
12    # Initialize and connect to the Streamable HTTP MCP server
13    mcp_tools = MCPTools(transport="streamable-http", url=server_url)
14    await mcp_tools.connect()
15
16    try:
17        agent = Agent(
18            model=OpenAIResponses(id="gpt-5.2"),
19            tools=[mcp_tools],
20            markdown=True,
21        )
22        await agent.aprint_response(message=message, stream=True, markdown=True)
23    finally:
24        await mcp_tools.close()
25
26
27# Using MultiMCPTools, we can connect to multiple MCP servers at once, even if they use different transports.
28# In this example we connect to both our example server (Streamable HTTP transport), and a different server (stdio transport).
29async def run_agent_with_multimcp(message: str) -> None:
30    # Initialize and connect to multiple MCP servers with different transports
31    mcp_tools = MultiMCPTools(
32        commands=["npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"],
33        urls=[server_url],
34        urls_transports=["streamable-http"],
35    )
36    await mcp_tools.connect()
37
38    try:
39        agent = Agent(
40            model=OpenAIResponses(id="gpt-5.2"),
41            tools=[mcp_tools],
42            markdown=True,
43        )
44        await agent.aprint_response(message=message, stream=True, markdown=True)
45    finally:
46        await mcp_tools.close()
47
48
49if __name__ == "__main__":
50    asyncio.run(run_agent("Do I have any birthdays this week?"))
51    asyncio.run(
52        run_agent_with_multimcp(
53            "Can you check when is my mom's birthday, and if there are any AirBnb listings in SF for two people for that day?"
54        )
55    )

Run the server

1python streamable_http_server.py

Run the client

1python streamable_http_client.py