Merge pull request #21 from modelcontextprotocol/davidsp/concepts

Python version of the concepts pages

docs/concepts/architecture.mdx

@@ -56,7 +56,37 @@ The protocol layer handles message framing, request/response linking, and high-l
   </Tab>
   <Tab title="Python">
     ```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
     ```
   </Tab>
 </Tabs>
@@ -212,7 +242,32 @@ Here's a basic example of implementing an MCP server:
   </Tab>
   <Tab title="Python">
     ```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)
     ```
   </Tab>
 </Tabs>

docs/concepts/prompts.mdx

@@ -292,7 +292,85 @@ Here's a complete example of implementing prompts in an MCP server:
   </Tab>
   <Tab title="Python">
     ```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")
     ```
   </Tab>
 </Tabs>

docs/concepts/resources.mdx

@@ -186,7 +186,33 @@ Here's a simple example of implementing resource support in an MCP server:
   </Tab>
   <Tab title="Python">
     ```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()
+        )
     ```
   </Tab>
 </Tabs>

docs/concepts/tools.mdx

@@ -82,7 +82,36 @@ Here's an example of implementing a basic tool in an MCP server:
   </Tab>
   <Tab title="Python">
     ```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}")
     ```
   </Tab>
 </Tabs>
@@ -212,6 +241,8 @@ Tool errors should be reported within the result object, not as MCP protocol-lev
 
 Here's an example of proper error handling for tools:
 
+<Tabs>
+  <Tab title="TypeScript">
     ```typescript
     try {
       // Tool operation
@@ -236,6 +267,33 @@ try {
       };
     }
     ```
+  </Tab>
+  <Tab title="Python">
+    ```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)}"
+                )
+            ]
+        )
+    ```
+  </Tab>
+</Tabs>
 
 This approach allows the LLM to see that an error occurred and potentially take corrective action or request human intervention.
 

docs/concepts/transports.mdx

@@ -59,9 +59,8 @@ Use stdio when:
 - Working with shell scripts
 
 <Tabs>
-  <Tab title="TypeScript">
+  <Tab title="TypeScript (Server)">
     ```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
+  </Tab>
+  <Tab title="TypeScript (Client)">
+    ```typescript
     const client = new Client({
       name: "example-client",
       version: "1.0.0"
@@ -83,9 +84,28 @@ Use stdio when:
     await client.connect(transport);
     ```
   </Tab>
-  <Tab title="Python">
+  <Tab title="Python (Server)">
     ```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()
+        )
+    ```
+  </Tab>
+  <Tab title="Python (Client)">
+    ```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()
     ```
   </Tab>
 </Tabs>
@@ -100,9 +120,8 @@ Use SSE when:
 - Implementing simple updates
 
 <Tabs>
-  <Tab title="TypeScript">
+  <Tab title="TypeScript (Server)">
     ```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
+  </Tab>
+  <Tab title="TypeScript (Client)">
+    ```typescript
     const client = new Client({
       name: "example-client",
       version: "1.0.0"
@@ -123,9 +144,35 @@ Use SSE when:
     await client.connect(transport);
     ```
   </Tab>
-  <Tab title="Python">
+  <Tab title="Python (Server)">
     ```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"]),
+        ]
+    )
+    ```
+  </Tab>
+  <Tab title="Python (Client)">
+    ```python
+    async with sse_client("http://localhost:8000/sse") as streams:
+        async with ClientSession(streams[0], streams[1]) as session:
+            await session.initialize()
     ```
   </Tab>
 </Tabs>
@@ -161,8 +208,38 @@ You can implement custom transports for:
     ```
   </Tab>
   <Tab title="Python">
+    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()
     ```
   </Tab>
 </Tabs>
@@ -204,8 +281,40 @@ Example error handling:
     ```
   </Tab>
   <Tab title="Python">
+  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
     ```
   </Tab>
 </Tabs>