Skip to content

MCP - Server HTTP Streamable Transport

server.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "mcp==1.21.0",
# ]
# ///

import argparse

from mcp.server.fastmcp import FastMCP  # type: ignore

mcp = FastMCP(
    name="Calculator",
    host="0.0.0.0",  # Only used for SSE transport (localhost)
    port=8333,  # Only used for SSE transport (set this to any port)
    stateless_http=True,
)


@mcp.tool()
def add(x: int, y: int) -> int:
    """Sum two integer numbers"""
    return x + y


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Run MCP server with optional transport"
    )
    parser.add_argument(
        "--transport",
        choices=["stdio", "sse", "streamable-http"],
        default="stdio",
        help="Transport method to use (default: stdio)",
    )
    args = parser.parse_args()

    print(f"Running server with {args.transport} transport")
    mcp.run(transport=args.transport)

Install packages

1
2
3
!uv pip install -q \
    mcp==1.21.0 \
    nest-asyncio==1.6.0

Import packages

1
2
3
4
5
6
7
import asyncio

import nest_asyncio  # type: ignore
from mcp import ClientSession  # type: ignore
from mcp.client.streamable_http import streamablehttp_client  # type: ignore

nest_asyncio.apply()  # Needed to run interactive python (Jupyter Notebooks)

Define client

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
async def main():
    async with streamablehttp_client("http://localhost:8333/mcp") as (
        read_stream,
        write_stream,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            available_tools = await session.list_tools()
            print("Available tools:")
            for tool in available_tools.tools:
                print(f"  - {tool.name}: {tool.description}")

            result = await session.call_tool("add", arguments={"x": 2, "y": 2})
            print(f"2 + 3 = {result.content[0].text}")

Run the server before running the client

1
uv run server.py --transport streamable-http

Run the client

1
2
if __name__ == "__main__":
    asyncio.run(main())

Available tools:

  - add: Sum two integer numbers

2 + 3 = 4