Closed
Description
Hi Guys , I want to build sse server and client from sdk example , but I'm not lucky, it can't work, SSE server and client got error :
SSE Server got starlette error like this :
.venv/lib/python3.13/site-packages/starlette/routing.py", line 74, in app
await response(scope, receive, send)
~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'NoneType' object is not callable
SSE Client ERROR: "Error in post_writer: " , anyone met have this issue ?
mcp 1.1.1
sse-starlette 2.1.3
starlette 0.41.3
uvicorn 0.32.1
sse server code :
import sys
import logging
import anyio
import click
import httpx
import mcp.types as types
from mcp.server import Server
logging.basicConfig(
#level=logging.CRITICAL,
level=logging.DEBUG,
format="%(asctime)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
async def fetch_website(
url: str,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
headers = {
"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
}
async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:
response = await client.get(url)
response.raise_for_status()
return [types.TextContent(type="text", text=response.text)]
@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option(
"--transport",
type=click.Choice(["stdio", "sse"]),
default="stdio",
help="Transport type",
)
def main(port: int, transport: str) -> int:
app = Server("mcp-website-fetcher")
@app.call_tool()
async def fetch_tool(
name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
if name != "fetch":
raise ValueError(f"Unknown tool: {name}")
if "url" not in arguments:
raise ValueError("Missing required argument 'url'")
return await fetch_website(arguments["url"])
@app.list_tools()
async def list_tools() -> list[types.Tool]:
print("list_tools")
return [
types.Tool(
name="fetch",
description="Fetches a website and returns its content",
inputSchema={
"type": "object",
"required": ["url"],
"properties": {
"url": {
"type": "string",
"description": "URL to fetch",
}
},
},
)
]
if transport == "sse":
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import Response
sse = SseServerTransport("/messages")
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)
async def handle_messages(request):
print(request)
await sse.handle_post_message(request.scope, request.receive, request._send)
async def handle_hello(request):
print(request)
return Response("Hello, World!", status_code=200)
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=handle_messages, methods=["POST"]),
Route("/hello", endpoint=handle_hello, methods=["GET"]),
],
)
import uvicorn
uvicorn.run(starlette_app, host="0.0.0.0", port=port)
else:
from mcp.server.stdio import stdio_server
async def arun():
async with stdio_server() as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)
anyio.run(arun)
return 0
main()sse client code:
import asyncio
import subprocess
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
def main():
# Start the server
try:
# Run the client logic
asyncio.run(client_logic())
finally:
# Terminate the server process
print("Server terminated.")
async def client_logic():
async with sse_client(url="http://127.0.0.1:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# List available tools
tools = await session.list_tools()
print(tools)
# Call the fetch tool
result = await session.call_tool("fetch", {"url": "https://example.com"})
print(result)
if __name__ == "__main__":
main()Metadata
Metadata
Assignees
Labels
No labels