diff --git a/docs/concepts/architecture.mdx b/docs/concepts/architecture.mdx
index 5e09797..5d85240 100644
--- a/docs/concepts/architecture.mdx
+++ b/docs/concepts/architecture.mdx
@@ -56,7 +56,37 @@ The protocol layer handles message framing, request/response linking, and high-l
```python
- # Python implementation coming soon
+ class Session(BaseSession[RequestT, NotificationT, ResultT]):
+ async def send_request(
+ self,
+ request: RequestT,
+ result_type: type[Result]
+ ) -> Result:
+ """
+ Send request and wait for response. Raises McpError if response contains error.
+ """
+ # Request handling implementation
+
+ async def send_notification(
+ self,
+ notification: NotificationT
+ ) -> None:
+ """Send one-way notification that doesn't expect response."""
+ # Notification handling implementation
+
+ async def _received_request(
+ self,
+ responder: RequestResponder[ReceiveRequestT, ResultT]
+ ) -> None:
+ """Handle incoming request from other side."""
+ # Request handling implementation
+
+ async def _received_notification(
+ self,
+ notification: ReceiveNotificationT
+ ) -> None:
+ """Handle incoming notification from other side."""
+ # Notification handling implementation
```
@@ -212,7 +242,32 @@ Here's a basic example of implementing an MCP server:
```python
- # Python implementation coming soon
+ import asyncio
+ import mcp.types as types
+ from mcp.server import Server
+ from mcp.server.stdio import stdio_server
+
+ app = Server("example-server")
+
+ @app.list_resources()
+ async def list_resources() -> list[types.Resource]:
+ return [
+ types.Resource(
+ uri="example://resource",
+ name="Example Resource"
+ )
+ ]
+
+ async def main():
+ async with stdio_server() as streams:
+ await app.run(
+ streams[0],
+ streams[1],
+ app.create_initialization_options()
+ )
+
+ if __name__ == "__main__":
+ asyncio.run(main)
```
diff --git a/docs/concepts/prompts.mdx b/docs/concepts/prompts.mdx
index d247ee8..91c005d 100644
--- a/docs/concepts/prompts.mdx
+++ b/docs/concepts/prompts.mdx
@@ -292,7 +292,85 @@ Here's a complete example of implementing prompts in an MCP server:
```python
- # Python implementation coming soon
+ from mcp.server import Server
+ import mcp.types as types
+
+ # Define available prompts
+ PROMPTS = {
+ "git-commit": types.Prompt(
+ name="git-commit",
+ description="Generate a Git commit message",
+ arguments=[
+ types.PromptArgument(
+ name="changes",
+ description="Git diff or description of changes",
+ required=True
+ )
+ ],
+ ),
+ "explain-code": types.Prompt(
+ name="explain-code",
+ description="Explain how code works",
+ arguments=[
+ types.PromptArgument(
+ name="code",
+ description="Code to explain",
+ required=True
+ ),
+ types.PromptArgument(
+ name="language",
+ description="Programming language",
+ required=False
+ )
+ ],
+ )
+ }
+
+ # Initialize server
+ app = Server("example-prompts-server")
+
+ @app.list_prompts()
+ async def list_prompts() -> list[types.Prompt]:
+ return list(PROMPTS.values())
+
+ @app.get_prompt()
+ async def get_prompt(
+ name: str, arguments: dict[str, str] | None = None
+ ) -> types.GetPromptResult:
+ if name not in PROMPTS:
+ raise ValueError(f"Prompt not found: {name}")
+
+ if name == "git-commit":
+ changes = arguments.get("changes") if arguments else ""
+ return types.GetPromptResult(
+ messages=[
+ types.PromptMessage(
+ role="user",
+ content=types.TextContent(
+ type="text",
+ text=f"Generate a concise but descriptive commit message "
+ f"for these changes:\n\n{changes}"
+ )
+ )
+ ]
+ )
+
+ if name == "explain-code":
+ code = arguments.get("code") if arguments else ""
+ language = arguments.get("language", "Unknown") if arguments else "Unknown"
+ return types.GetPromptResult(
+ messages=[
+ types.PromptMessage(
+ role="user",
+ content=types.TextContent(
+ type="text",
+ text=f"Explain how this {language} code works:\n\n{code}"
+ )
+ )
+ ]
+ )
+
+ raise ValueError("Prompt implementation not found")
```
diff --git a/docs/concepts/resources.mdx b/docs/concepts/resources.mdx
index 00f6436..d61722a 100644
--- a/docs/concepts/resources.mdx
+++ b/docs/concepts/resources.mdx
@@ -186,7 +186,33 @@ Here's a simple example of implementing resource support in an MCP server:
```python
- # Python implementation coming soon
+ app = Server("example-server")
+
+ @app.list_resources()
+ async def list_resources() -> list[types.Resource]:
+ return [
+ types.Resource(
+ uri="file:///logs/app.log",
+ name="Application Logs",
+ mimeType="text/plain"
+ )
+ ]
+
+ @app.read_resource()
+ async def read_resource(uri: AnyUrl) -> str:
+ if str(uri) == "file:///logs/app.log":
+ log_contents = await read_log_file()
+ return log_contents
+
+ raise ValueError("Resource not found")
+
+ # Start server
+ async with stdio_server() as streams:
+ await app.run(
+ streams[0],
+ streams[1],
+ app.create_initialization_options()
+ )
```
diff --git a/docs/concepts/tools.mdx b/docs/concepts/tools.mdx
index 363b9c1..17cc10c 100644
--- a/docs/concepts/tools.mdx
+++ b/docs/concepts/tools.mdx
@@ -82,7 +82,36 @@ Here's an example of implementing a basic tool in an MCP server:
```python
- # Python implementation coming soon
+ app = Server("example-server")
+
+ @app.list_tools()
+ async def list_tools() -> list[types.Tool]:
+ return [
+ types.Tool(
+ name="calculate_sum",
+ description="Add two numbers together",
+ inputSchema={
+ "type": "object",
+ "properties": {
+ "a": {"type": "number"},
+ "b": {"type": "number"}
+ },
+ "required": ["a", "b"]
+ }
+ )
+ ]
+
+ @app.call_tool()
+ async def call_tool(
+ name: str,
+ arguments: dict
+ ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
+ if name == "calculate_sum":
+ a = arguments["a"]
+ b = arguments["b"]
+ result = a + b
+ return [types.TextContent(type="text", text=str(result))]
+ raise ValueError(f"Tool not found: {name}")
```
@@ -212,30 +241,59 @@ Tool errors should be reported within the result object, not as MCP protocol-lev
Here's an example of proper error handling for tools:
-```typescript
-try {
- // Tool operation
- const result = performOperation();
- return {
- content: [
- {
- type: "text",
- text: `Operation successful: ${result}`
- }
- ]
- };
-} catch (error) {
- return {
- isError: true,
- content: [
- {
- type: "text",
- text: `Error: ${error.message}`
- }
- ]
- };
-}
-```
+
+
+ ```typescript
+ try {
+ // Tool operation
+ const result = performOperation();
+ return {
+ content: [
+ {
+ type: "text",
+ text: `Operation successful: ${result}`
+ }
+ ]
+ };
+ } catch (error) {
+ return {
+ isError: true,
+ content: [
+ {
+ type: "text",
+ text: `Error: ${error.message}`
+ }
+ ]
+ };
+ }
+ ```
+
+
+ ```python
+ try:
+ # Tool operation
+ result = perform_operation()
+ return types.CallToolResult(
+ content=[
+ types.TextContent(
+ type="text",
+ text=f"Operation successful: {result}"
+ )
+ ]
+ )
+ except Exception as error:
+ return types.CallToolResult(
+ isError=True,
+ content=[
+ types.TextContent(
+ type="text",
+ text=f"Error: {str(error)}"
+ )
+ ]
+ )
+ ```
+
+
This approach allows the LLM to see that an error occurred and potentially take corrective action or request human intervention.
diff --git a/docs/concepts/transports.mdx b/docs/concepts/transports.mdx
index 3c0fd89..a71810b 100644
--- a/docs/concepts/transports.mdx
+++ b/docs/concepts/transports.mdx
@@ -59,9 +59,8 @@ Use stdio when:
- Working with shell scripts
-
+
```typescript
- // Server implementation
const server = new Server({
name: "example-server",
version: "1.0.0"
@@ -69,8 +68,10 @@ Use stdio when:
const transport = new StdioServerTransport();
await server.connect(transport);
-
- // Client implementation
+ ```
+
+
+ ```typescript
const client = new Client({
name: "example-client",
version: "1.0.0"
@@ -83,9 +84,28 @@ Use stdio when:
await client.connect(transport);
```
-
+
```python
- # Python implementation coming soon
+ app = Server("example-server")
+
+ async with stdio_server() as streams:
+ await app.run(
+ streams[0],
+ streams[1],
+ app.create_initialization_options()
+ )
+ ```
+
+
+ ```python
+ params = StdioServerParameters(
+ command="./server",
+ args=["--option", "value"]
+ )
+
+ async with stdio_client(params) as streams:
+ async with ClientSession(streams[0], streams[1]) as session:
+ await session.initialize()
```
@@ -100,9 +120,8 @@ Use SSE when:
- Implementing simple updates
-
+
```typescript
- // Server implementation
const server = new Server({
name: "example-server",
version: "1.0.0"
@@ -110,8 +129,10 @@ Use SSE when:
const transport = new SSEServerTransport("/message", response);
await server.connect(transport);
-
- // Client implementation
+ ```
+
+
+ ```typescript
const client = new Client({
name: "example-client",
version: "1.0.0"
@@ -123,9 +144,35 @@ Use SSE when:
await client.connect(transport);
```
-
+
```python
- # Python implementation coming soon
+ from mcp.server.sse import SseServerTransport
+ from starlette.applications import Starlette
+ from starlette.routing import Route
+
+ app = Server("example-server")
+ sse = SseServerTransport("/messages")
+
+ async def handle_sse(scope, receive, send):
+ async with sse.connect_sse(scope, receive, send) as streams:
+ await app.run(streams[0], streams[1], app.create_initialization_options())
+
+ async def handle_messages(scope, receive, send):
+ await sse.handle_post_message(scope, receive, send)
+
+ starlette_app = Starlette(
+ routes=[
+ Route("/sse", endpoint=handle_sse),
+ Route("/messages", endpoint=handle_messages, methods=["POST"]),
+ ]
+ )
+ ```
+
+
+ ```python
+ async with sse_client("http://localhost:8000/sse") as streams:
+ async with ClientSession(streams[0], streams[1]) as session:
+ await session.initialize()
```
@@ -161,8 +208,38 @@ You can implement custom transports for:
```
+ Note that while MCP Servers are often implemented with asyncio, we recommend
+ implementing low-level interfaces like transports with `anyio` for wider compatibility.
```python
- # Python implementation coming soon
+ @contextmanager
+ async def create_transport(
+ read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
+ write_stream: MemoryObjectSendStream[JSONRPCMessage]
+ ):
+ """
+ Transport interface for MCP.
+
+ Args:
+ read_stream: Stream to read incoming messages from
+ write_stream: Stream to write outgoing messages to
+ """
+ async with anyio.create_task_group() as tg:
+ try:
+ # Start processing messages
+ tg.start_soon(lambda: process_messages(read_stream))
+
+ # Send messages
+ async with write_stream:
+ yield write_stream
+
+ except Exception as exc:
+ # Handle errors
+ raise exc
+ finally:
+ # Clean up
+ tg.cancel_scope.cancel()
+ await write_stream.aclose()
+ await read_stream.aclose()
```
@@ -204,8 +281,40 @@ Example error handling:
```
+ Note that while MCP Servers are often implemented with asyncio, we recommend
+ implementing low-level interfaces like transports with `anyio` for wider compatibility.
```python
- # Python implementation coming soon
+ @contextmanager
+ async def example_transport(scope: Scope, receive: Receive, send: Send):
+ try:
+ # Create streams for bidirectional communication
+ read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
+ write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
+
+ async def message_handler():
+ try:
+ async with read_stream_writer:
+ # Message handling logic
+ pass
+ except Exception as exc:
+ logger.error(f"Failed to handle message: {exc}")
+ raise exc
+
+ async with anyio.create_task_group() as tg:
+ tg.start_soon(message_handler)
+ try:
+ # Yield streams for communication
+ yield read_stream, write_stream
+ except Exception as exc:
+ logger.error(f"Transport error: {exc}")
+ raise exc
+ finally:
+ tg.cancel_scope.cancel()
+ await write_stream.aclose()
+ await read_stream.aclose()
+ except Exception as exc:
+ logger.error(f"Failed to initialize transport: {exc}")
+ raise exc
```