mirror of
https://github.com/aljazceru/mcp-python-sdk.git
synced 2025-12-18 14:34:27 +01:00
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
import anyio
|
|
import click
|
|
import mcp.types as types
|
|
from mcp.server.lowlevel import Server
|
|
from mcp.shared._httpx_utils import create_mcp_http_client
|
|
|
|
|
|
async def fetch_website(
|
|
url: str,
|
|
) -> list[types.Content]:
|
|
headers = {
|
|
"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
|
|
}
|
|
async with create_mcp_http_client(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.Content]:
|
|
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]:
|
|
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.responses import Response
|
|
from starlette.routing import Mount, Route
|
|
|
|
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()
|
|
)
|
|
return Response()
|
|
|
|
starlette_app = Starlette(
|
|
debug=True,
|
|
routes=[
|
|
Route("/sse", endpoint=handle_sse, methods=["GET"]),
|
|
Mount("/messages/", app=sse.handle_post_message),
|
|
],
|
|
)
|
|
|
|
import uvicorn
|
|
|
|
uvicorn.run(starlette_app, host="127.0.0.1", 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
|