SSE Transport

Kern'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.

Note

This transport is not recommended anymore by the MCP protocol. Use the Streamable HTTP transport instead.

To use it, initialize the MCPTools passing the URL of the MCP server and setting the transport to sse:

1from kern.agent import Agent
2from kern.models.openai import OpenAIResponses
3from kern.tools.mcp import MCPTools
4
5server_url = "http://localhost:8000/sse"
6
7# Initialize and connect to the SSE MCP server
8mcp_tools = MCPTools(url=server_url, transport="sse")
9await mcp_tools.connect()
10
11try:
12    agent = Agent(model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools])
13    await agent.aprint_response("What is the license for this project?", stream=True)
14finally:
15    # Always close the connection when done
16    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, SSEClientParams
2
3server_params = SSEClientParams(
4    url=...,
5    headers=...,
6    timeout=...,
7    sse_read_timeout=...,
8)
9
10# Initialize and connect using server parameters
11mcp_tools = MCPTools(server_params=server_params, transport="sse")
12await mcp_tools.connect()
13
14try:
15    # Use mcp_tools with your agent
16    pass
17finally:
18    await mcp_tools.close()

Complete example

Let's set up a simple local server and connect to it using the SSE 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="sse")

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/sse"
9
10
11async def run_agent(message: str) -> None:
12    # Initialize and connect to the SSE MCP server
13    mcp_tools = MCPTools(transport="sse", 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 (SSE 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=["sse"],
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 sse_server.py

Run the client

1python sse_client.py