Merge pull request #696 from realpython/python_mcp

completed mcp code

Author: brendaweles

python-mcp/README.md

@@ -0,0 +1,23 @@
+# Python MCP: Connect Your LLM With the World
+
+This folder contains the source code for [Python MCP: Connect Your LLM With the World](https://realpython.com/python-mcp-connect-your-llm-with-world/).
+
+## Setup
+
+Create a new virtual environment, and run the following command to install Python MCP and the additional requirements for this project:
+
+```console
+(venv) $ python -m pip install "mcp[cli]"
+```
+
+You'll use [`pytest-asyncio`](https://realpython.com/pytest-python-testing/) to test your MCP server. With that, you've installed all the Python dependencies you'll need for this tutorial, and you're ready to dive into MCP!
+
+## Usage
+
+Once your environment is set up, you can run the MCP server with the following command:
+
+```console
+(venv) $ python main.py
+```
+
+In `test/test_server.py`, you'll also want to change `SERVER_PATH` to the local path to your `main.py` file. See the tutorial for all the details on what's going on here, along with how to connect your server to Cursor's MCP client.

python-mcp/main.py

@@ -0,0 +1,82 @@
+import asyncio
+
+from mcp.server.fastmcp import FastMCP
+from transactional_db import CUSTOMERS_TABLE, ORDERS_TABLE, PRODUCTS_TABLE
+
+mcp = FastMCP("ecommerce_tools")
+
+
+@mcp.tool()
+async def get_customer_info(customer_id: str) -> str:
+    """Search for a customer using their unique identifier"""
+    await asyncio.sleep(1)
+    customer_info = CUSTOMERS_TABLE.get(customer_id)
+
+    if not customer_info:
+        return "Customer not found"
+
+    return str(customer_info)
+
+
+@mcp.tool()
+async def get_order_details(order_id: str) -> str:
+    """Get details about a specific order."""
+    await asyncio.sleep(1)
+    order = ORDERS_TABLE.get(order_id)
+    if not order:
+        return f"No order found with ID {order_id}."
+
+    items = [
+        PRODUCTS_TABLE[sku]["name"]
+        for sku in order["items"]
+        if sku in PRODUCTS_TABLE
+    ]
+    return (
+        f"Order ID: {order_id}\n"
+        f"Customer ID: {order['customer_id']}\n"
+        f"Date: {order['date']}\n"
+        f"Status: {order['status']}\n"
+        f"Total: ${order['total']:.2f}\n"
+        f"Items: {', '.join(items)}"
+    )
+
+
+@mcp.tool()
+async def check_inventory(product_name: str) -> str:
+    """Search inventory for a product by product name."""
+    await asyncio.sleep(1)
+    matches = []
+    for sku, product in PRODUCTS_TABLE.items():
+        if product_name.lower() in product["name"].lower():
+            matches.append(
+                f"{product['name']} (SKU: {sku}) — Stock: {product['stock']}"
+            )
+    return "\n".join(matches) if matches else "No matching products found."
+
+
+@mcp.tool()
+async def get_customer_ids_by_name(customer_name: str) -> list[str]:
+    """Get customer IDs by using a customer's full name"""
+    await asyncio.sleep(1)
+    return [
+        cust_id
+        for cust_id, info in CUSTOMERS_TABLE.items()
+        if info.get("name") == customer_name
+    ]
+
+
+@mcp.tool()
+async def get_orders_by_customer_id(
+    customer_id: str,
+) -> dict[str, dict[str, str]]:
+    """Get orders by customer ID"""
+    await asyncio.sleep(1)
+    return {
+        order_id: order
+        for order_id, order in ORDERS_TABLE.items()
+        if order.get("customer_id") == customer_id
+    }
+
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

python-mcp/pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "python-mcp"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "httpx>=0.28.1",
+    "mcp[cli]>=1.7.1",
+    "pypdf2>=3.0.1",
+    "pytest-asyncio>=0.26.0",
+]

python-mcp/tests/test_server.py

@@ -0,0 +1,46 @@
+from contextlib import AsyncExitStack
+
+import pytest
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+SERVER_PATH = "/path/to/your/python-mcp/main.py"
+EXPECTED_TOOLS = [
+    "get_customer_info",
+    "get_order_details",
+    "check_inventory",
+    "get_customer_ids_by_name",
+    "get_orders_by_customer_id",
+]
+
+
+@pytest.mark.asyncio
+async def test_mcp_server_connection():
+    """Connect to an MCP server and verify the tools"""
+
+    exit_stack = AsyncExitStack()
+
+    server_params = StdioServerParameters(
+        command="python", args=[SERVER_PATH], env=None
+    )
+
+    stdio_transport = await exit_stack.enter_async_context(
+        stdio_client(server_params)
+    )
+    stdio, write = stdio_transport
+    session = await exit_stack.enter_async_context(ClientSession(stdio, write))
+
+    await session.initialize()
+
+    response = await session.list_tools()
+    tools = response.tools
+    tool_names = [tool.name for tool in tools]
+    tool_descriptions = [tool.description for tool in tools]
+
+    print("\nYour server has the follow tools:")
+    for tool_name, tool_description in zip(tool_names, tool_descriptions):
+        print(f"{tool_name}: {tool_description}")
+
+    assert sorted(EXPECTED_TOOLS) == sorted(tool_names)
+
+    await exit_stack.aclose()

python-mcp/transactional_db.py

@@ -0,0 +1,42 @@
+CUSTOMERS_TABLE = {
+    "CUST123": {
+        "name": "Alice Johnson",
+        "email": "[email protected]",
+        "phone": "555-1234",
+    },
+    "CUST456": {
+        "name": "Bob Smith",
+        "email": "[email protected]",
+        "phone": "555-5678",
+    },
+}
+
+ORDERS_TABLE = {
+    "ORD1001": {
+        "customer_id": "CUST123",
+        "date": "2024-04-01",
+        "status": "Shipped",
+        "total": 89.99,
+        "items": ["SKU100", "SKU200"],
+    },
+    "ORD1015": {
+        "customer_id": "CUST123",
+        "date": "2024-05-17",
+        "status": "Processing",
+        "total": 45.50,
+        "items": ["SKU300"],
+    },
+    "ORD1022": {
+        "customer_id": "CUST456",
+        "date": "2024-06-04",
+        "status": "Delivered",
+        "total": 120.00,
+        "items": ["SKU100", "SKU100"],
+    },
+}
+
+PRODUCTS_TABLE = {
+    "SKU100": {"name": "Wireless Mouse", "price": 29.99, "stock": 42},
+    "SKU200": {"name": "Keyboard", "price": 59.99, "stock": 18},
+    "SKU300": {"name": "USB-C Cable", "price": 15.50, "stock": 77},
+}