Say MCP Server
by bmorphism
- mcp_simple_tool
import anyio
import click
import httpx
import mcp.types as types
from mcp.server import Server
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]:
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
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):
await sse.handle_post_message(request.scope, request.receive, request._send)
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=handle_messages, methods=["POST"]),
],
)
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