mcp_server_quickstart_with_resources_and_echo_tool.py

1import asyncio
2
3from mcp.server.models import InitializationOptions
4import mcp.types as types
5from mcp.server import NotificationOptions, Server
6from mcp.server.stdio import stdio_server
7
8# Initialize the server
9server = Server("example-server")
10
11# Define a simple resource
12@server.list_resources()
13async def handle_list_resources() -> list[types.Resource]:
14    return [
15        types.Resource(
16            uri="file:///example.txt",
17            name="Example Text",
18            description="An example text file",
19            mimeType="text/plain",
20        )
21    ]
22
23@server.read_resource()
24async def handle_read_resource(uri: str) -> str:
25    if uri == "file:///example.txt":
26        return "This is the content of the example file."
27    raise ValueError(f"Resource not found: {uri}")
28
29# Define a tool
30@server.list_tools()
31async def handle_list_tools() -> list[types.Tool]:
32    return [
33        types.Tool(
34            name="echo",
35            description="Echoes back the input",
36            inputSchema={
37                "type": "object",
38                "properties": {
39                    "message": {"type": "string"},
40                },
41                "required": ["message"],
42            },
43        )
44    ]
45
46@server.call_tool()
47async def handle_call_tool(
48    name: str, arguments: dict | None
49) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
50    if name == "echo":
51        message = arguments.get("message", "")
52        return [types.TextContent(type="text", text=f"Echo: {message}")]
53    raise ValueError(f"Unknown tool: {name}")
54
55async def main():
56    # Run the server using stdin/stdout streams
57    async with stdio_server() as (read_stream, write_stream):
58        await server.run(
59            read_stream,
60            write_stream,
61            InitializationOptions(
62                server_name="example-server",
63                server_version="0.1.0",
64                capabilities=server.get_capabilities(
65                    notification_options=NotificationOptions(),
66                    experimental_capabilities={},
67                ),
68            ),
69        )
70
71if __name__ == "__main__":
72    asyncio.run(main())