From f1fef38b4bbfd25ee9cc7bbc2da1d93f113a99c3 Mon Sep 17 00:00:00 2001 From: Farid Dihan Nahdi Date: Mon, 12 Jan 2026 04:30:24 +0700 Subject: [PATCH] feat: Add MCP (Model Context Protocol) sample implementation This commit adds a new MCP binding sample for UCP, enabling AI agents to interact with UCP-compliant merchants through the Model Context Protocol. New files: - mcp/python/ucp_mcp_server.py: FastMCP server exposing UCP capabilities - Tools: list_products, create_checkout, add_to_checkout, complete_payment, etc. - Resources: catalog, checkout sessions, orders, discovery profile - Prompts: shopping assistant, order confirmation, product recommendations - Supports stdio, HTTP, and SSE transports - mcp/python/ucp_mcp_client.py: Example client demonstrating complete shopping flow - Connects via MCP protocol - Demonstrates happy path: browse -> checkout -> pay -> track - mcp/python/README.md: Comprehensive documentation - Architecture diagram - Quick start guide - Claude Desktop integration - Development guide - mcp/python/pyproject.toml: Project configuration with dependencies Updated: - README.md: Added MCP binding section with comparison table This sample enables conversational commerce experiences by allowing LLMs like Claude, GPT, and Gemini to perform shopping operations through the standardized MCP protocol. --- README.md | 41 +- mcp/python/README.md | 363 +++++++++++++ mcp/python/pyproject.toml | 48 ++ mcp/python/ucp_mcp_client.py | 348 +++++++++++++ mcp/python/ucp_mcp_server.py | 980 +++++++++++++++++++++++++++++++++++ 5 files changed, 1778 insertions(+), 2 deletions(-) create mode 100644 mcp/python/README.md create mode 100644 mcp/python/pyproject.toml create mode 100644 mcp/python/ucp_mcp_client.py create mode 100644 mcp/python/ucp_mcp_server.py diff --git a/README.md b/README.md index 61eaa64..ef2177e 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,9 @@ Universal Commerce Protocol (UCP). ## Sample Implementations -### Python +### REST Bindings + +#### Python (REST) A reference implementation of a UCP Merchant Server using Python and FastAPI. @@ -39,7 +41,7 @@ A reference implementation of a UCP Merchant Server using Python and FastAPI. * A script demonstrating a full "happy path" user journey (discovery -> checkout -> payment). -### Node.js +#### Node.js (REST) A reference implementation of a UCP Merchant Server using Node.js, Hono, and Zod. @@ -49,6 +51,41 @@ Zod. * Demonstrates implementation of UCP specifications for shopping, checkout, and order management using a Node.js stack. +### MCP Binding (Model Context Protocol) + +A reference implementation using the [Model Context Protocol](https://modelcontextprotocol.io/) +for AI agent integration. + +* **Server & Client**: [Documentation](mcp/python/README.md) + + * Located in `mcp/python/`. + * Exposes UCP shopping capabilities as MCP tools and resources. + * Enables AI agents (Claude, GPT, etc.) to perform commerce operations. + * Includes example client demonstrating complete shopping flow. + +**Key Features:** +* Tools for product discovery, checkout, payment, and order management +* Resources for read-only access to catalogs, sessions, and orders +* Prompts for reusable conversation patterns +* Support for stdio, HTTP, and SSE transports + +### A2A Binding (Agent-to-Agent) + +An example Business A2A Agent implementing the UCP Extension. + +* **Agent & Client**: [Documentation](a2a/README.md) + * Located in `a2a/`. + * Demonstrates agent-to-agent communication for commerce operations. + * Includes chat client for interactive testing. + +## Protocol Binding Comparison + +| Binding | Use Case | Transport | Best For | +|---------|----------|-----------|----------| +| REST | Traditional APIs | HTTP | Web/mobile apps, service integration | +| MCP | AI Agents | stdio/HTTP/SSE | LLM integration, conversational commerce | +| A2A | Agent Communication | HTTP | Multi-agent systems, automated workflows | + ## Getting Started Please refer to the specific README files linked above for detailed instructions diff --git a/mcp/python/README.md b/mcp/python/README.md new file mode 100644 index 0000000..346553f --- /dev/null +++ b/mcp/python/README.md @@ -0,0 +1,363 @@ + + +# UCP MCP Sample (Python) + +This is a reference implementation of a UCP Merchant Server using the +[Model Context Protocol (MCP)](https://modelcontextprotocol.io/). It demonstrates +how to expose UCP shopping capabilities as MCP tools and resources, enabling +AI agents to perform commerce operations through a standardized protocol. + +## Overview + +The Model Context Protocol (MCP) is an open standard for connecting AI systems +with external data sources and tools. This sample shows how UCP's shopping +capabilities can be exposed via MCP, allowing LLMs and AI agents to: + +- Browse product catalogs +- Manage checkout sessions +- Process payments +- Track orders + +### Why MCP for UCP? + +| Feature | REST API | MCP | +|---------|----------|-----| +| Integration | HTTP endpoints, OpenAPI | Standardized protocol | +| AI Compatibility | Requires custom tooling | Native LLM integration | +| Discovery | `/.well-known/ucp` | Built-in capability listing | +| Interactivity | Request/Response | Bidirectional, streaming | +| Context | Stateless | Session-aware | + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ AI Agent / LLM │ +│ (Claude, GPT, Gemini, etc.) │ +└─────────────────────────────┬───────────────────────────────────┘ + │ MCP Protocol + │ (stdio / HTTP / SSE) + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ UCP MCP Server │ +│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ +│ │ Tools │ │ Resources │ │ Prompts │ │ +│ │ ─────────────── │ │ ─────────────── │ │ ─────────────── │ │ +│ │ • list_products │ │ • catalog │ │ • shopping_intro│ │ +│ │ • get_product │ │ • checkout/{id} │ │ • order_confirm │ │ +│ │ • create_checkout│ │ • orders/{id} │ │ • recommend │ │ +│ │ • add_to_checkout│ │ • discovery │ │ │ │ +│ │ • set_shipping │ │ │ │ │ │ +│ │ • complete_pay │ │ │ │ │ │ +│ │ • get_order │ │ │ │ │ │ +│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ +└─────────────────────────────┬───────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Mock Data Store / Real UCP Server │ +│ (Products, Checkout, Orders) │ +└─────────────────────────────────────────────────────────────────┘ +``` + +## Prerequisites + +- Python 3.11 or later +- [uv](https://docs.astral.sh/uv/) package manager + +## Quick Start + +### 1. Install Dependencies + +```bash +cd mcp/python +uv sync +``` + +### 2. Run the MCP Server + +**Using stdio transport (for LLM integration):** + +```bash +uv run ucp-mcp-server +``` + +**Using HTTP transport (for web clients):** + +```bash +uv run ucp-mcp-server --transport http --port 8000 +``` + +The server will start and expose UCP capabilities via the MCP protocol. + +### 3. Run the Example Client + +In a separate terminal, run the demo client that performs a complete shopping +flow: + +```bash +uv run ucp-mcp-client +``` + +This demonstrates: +1. Connecting to the MCP server +2. Listing available products +3. Creating a checkout session +4. Adding items to cart +5. Setting shipping address +6. Completing payment +7. Retrieving order confirmation + +## Project Structure + +``` +mcp/python/ +├── pyproject.toml # Project configuration and dependencies +├── README.md # This documentation +├── ucp_mcp_server.py # MCP server implementation +└── ucp_mcp_client.py # Example client demonstrating usage +``` + +## Server Components + +### Tools (Actions with Side Effects) + +Tools are functions that can modify state, similar to POST/PUT/DELETE endpoints: + +| Tool | Description | +|------|-------------| +| `list_products` | List available products, optionally filtered by category | +| `get_product` | Get detailed information about a specific product | +| `create_checkout` | Create a new checkout session | +| `add_to_checkout` | Add a product to an existing checkout | +| `remove_from_checkout` | Remove a product from checkout | +| `set_shipping_address` | Set the delivery address | +| `get_checkout` | Get current checkout state | +| `complete_payment` | Process payment and create order | +| `get_order` | Retrieve order details and status | +| `cancel_checkout` | Cancel an open checkout session | + +### Resources (Read-Only Data) + +Resources provide read-only access to data, similar to GET endpoints: + +| Resource URI | Description | +|--------------|-------------| +| `ucp://catalog/products` | Complete product catalog | +| `ucp://catalog/products/{id}` | Specific product details | +| `ucp://checkout/{session_id}` | Checkout session state | +| `ucp://orders/{order_id}` | Order details and tracking | +| `ucp://discovery/profile` | UCP discovery profile | + +### Prompts (Conversation Templates) + +Prompts provide reusable conversation patterns for AI agents: + +| Prompt | Description | +|--------|-------------| +| `shopping_intro` | Introduction for shopping assistant | +| `order_confirmation` | Order confirmation message template | +| `recommend_products` | Product recommendations for occasions | + +## Integration with Claude Desktop + +To use this MCP server with Claude Desktop, add the following to your +`claude_desktop_config.json`: + +```json +{ + "mcpServers": { + "ucp-shopping": { + "command": "uv", + "args": ["run", "ucp-mcp-server"], + "cwd": "/path/to/samples/mcp/python" + } + } +} +``` + +After restarting Claude Desktop, you can ask Claude to help you shop: + +> "I'm looking for flowers for Mother's Day. Can you help me find something nice?" + +Claude will use the MCP tools to browse products and guide you through checkout. + +## Integration with Other LLM Clients + +### Using HTTP Transport + +Start the server with HTTP transport: + +```bash +uv run ucp-mcp-server --transport http --port 8000 +``` + +Then connect your client to `http://localhost:8000/mcp`. + +### Using SSE Transport + +For Server-Sent Events (legacy compatibility): + +```bash +uv run ucp-mcp-server --transport sse --port 8000 +``` + +## Example Interaction + +Here's an example of how an AI agent might use the MCP tools: + +``` +User: I'd like to order some flowers for a birthday + +Agent: I'd be happy to help you find flowers for a birthday! Let me show you + what we have available. + + [Calls list_products()] + + We have these beautiful options: + 1. Red Roses Bouquet - $49.99 + 2. Mixed Tulips - $34.99 + 3. Sunflower Arrangement - $44.99 + 4. Stargazer Lilies - $54.99 + + Which would you like to add to your cart? + +User: Let's go with the stargazer lilies + +Agent: Great choice! Stargazer lilies have a beautiful fragrance. + + [Calls create_checkout()] + [Calls add_to_checkout(product_id="prod_lily_stargazer", quantity=1)] + + I've added 1 Stargazer Lilies to your cart. + + Current total: $54.99 + $4.40 tax = $59.39 + + Would you like to add anything else, or shall we proceed to checkout? + +User: That's all. Ship it to 456 Oak Ave, Seattle WA 98101 + +Agent: [Calls set_shipping_address(...)] + + Your shipping address is set. Here's your order summary: + + Items: 1x Stargazer Lilies - $54.99 + Tax: $4.40 + Shipping: $9.99 + Total: $69.38 + + Ready to complete your purchase? + +User: Yes, please complete the order + +Agent: [Calls complete_payment()] + + 🎉 Order placed successfully! + + Order ID: order_a1b2c3d4e5f6 + Tracking: TRKABCDEF1234 + + You'll receive a confirmation email shortly. The flowers will be + delivered within 1-2 business days. +``` + +## Development + +### Running Tests + +```bash +uv run pytest +``` + +### Adding New Tools + +To add a new tool, use the `@mcp.tool()` decorator: + +```python +@mcp.tool() +async def my_new_tool( + param1: str, + param2: int = 10, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Description of what the tool does. + + Args: + param1: Description of first parameter. + param2: Description of optional parameter (default: 10). + + Returns: + Dictionary with result data. + """ + if ctx: + await ctx.info("Performing operation...") + + # Implementation + return {"result": "success"} +``` + +### Adding New Resources + +To add a new resource, use the `@mcp.resource()` decorator: + +```python +@mcp.resource("ucp://my-resource/{param}") +def my_resource(param: str) -> str: + """ + Description of the resource. + + Returns JSON data for the specified parameter. + """ + data = {"param": param, "info": "..."} + return json.dumps(data, indent=2) +``` + +## Connecting to Real UCP Server + +This sample includes a mock data store for standalone testing. To connect to +a real UCP REST server, you would modify the tool implementations to make HTTP +calls: + +```python +import httpx + +UCP_SERVER_URL = "http://localhost:8182" + +@mcp.tool() +async def list_products(category: str | None = None) -> dict: + async with httpx.AsyncClient() as client: + response = await client.get( + f"{UCP_SERVER_URL}/products", + params={"category": category} if category else None + ) + return response.json() +``` + +## Related Resources + +- [UCP Specification](https://github.com/Universal-Commerce-Protocol/ucp) +- [MCP Documentation](https://modelcontextprotocol.io/) +- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) +- [REST Sample](../rest/python/server/README.md) +- [A2A Sample](../a2a/README.md) + +## Disclaimer + +This is an example implementation for demonstration purposes and is not +intended for production use without additional security, error handling, +and integration with real payment providers. diff --git a/mcp/python/pyproject.toml b/mcp/python/pyproject.toml new file mode 100644 index 0000000..1b3f563 --- /dev/null +++ b/mcp/python/pyproject.toml @@ -0,0 +1,48 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +[project] +name = "ucp-mcp-sample" +version = "0.1.0" +description = "UCP MCP (Model Context Protocol) Sample Implementation" +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.11" +dependencies = [ + "mcp>=1.1.0", + "pydantic>=2.0.0", + "httpx>=0.27.0", + "uvicorn>=0.32.0", + "starlette>=0.41.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.24.0", +] + +[project.scripts] +ucp-mcp-server = "ucp_mcp_server:main" +ucp-mcp-client = "ucp_mcp_client:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["."] + +[tool.pytest.ini_options] +asyncio_mode = "auto" diff --git a/mcp/python/ucp_mcp_client.py b/mcp/python/ucp_mcp_client.py new file mode 100644 index 0000000..8d55331 --- /dev/null +++ b/mcp/python/ucp_mcp_client.py @@ -0,0 +1,348 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +UCP MCP Client - Example client demonstrating MCP interaction with UCP server. + +This module demonstrates how to connect to the UCP MCP server and perform +shopping operations using the MCP protocol. It showcases a complete "happy path" +user journey: + +1. Connect to the MCP server +2. List available products +3. Create a checkout session +4. Add items to checkout +5. Set shipping address +6. Complete payment +7. Retrieve order confirmation + +Usage: + # Connect to a running MCP server via stdio: + uv run ucp-mcp-client + + # Connect via HTTP: + uv run ucp-mcp-client --transport http --url http://localhost:8000/mcp +""" + +import argparse +import asyncio +import json +import logging +from typing import Any + +from mcp import ClientSession +from mcp.client.stdio import StdioServerParameters, stdio_client +from mcp.types import TextContent + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +async def print_separator(title: str): + """Print a visual separator with title.""" + print(f"\n{'='*60}") + print(f" {title}") + print(f"{'='*60}\n") + + +async def display_tools(session: ClientSession): + """Display available tools from the server.""" + await print_separator("Available Tools") + + tools = await session.list_tools() + for tool in tools.tools: + print(f" 📦 {tool.name}") + if tool.description: + # Show first line of description + desc = tool.description.split('\n')[0].strip() + print(f" {desc}") + + +async def display_resources(session: ClientSession): + """Display available resources from the server.""" + await print_separator("Available Resources") + + # List static resources + resources = await session.list_resources() + for resource in resources.resources: + print(f" 📄 {resource.uri}") + if resource.name: + print(f" Name: {resource.name}") + + # List resource templates + templates = await session.list_resource_templates() + for template in templates.resourceTemplates: + print(f" 📋 {template.uriTemplate} (template)") + if template.name: + print(f" Name: {template.name}") + + +async def display_prompts(session: ClientSession): + """Display available prompts from the server.""" + await print_separator("Available Prompts") + + prompts = await session.list_prompts() + for prompt in prompts.prompts: + print(f" 💬 {prompt.name}") + if prompt.description: + print(f" {prompt.description}") + + +async def call_tool(session: ClientSession, name: str, arguments: dict[str, Any]) -> dict: + """Call a tool and return the result as a dictionary.""" + logger.info(f"Calling tool: {name} with args: {arguments}") + + result = await session.call_tool(name, arguments=arguments) + + # Extract text content from result + if result.content and len(result.content) > 0: + content = result.content[0] + if isinstance(content, TextContent): + try: + return json.loads(content.text) + except json.JSONDecodeError: + return {"text": content.text} + + # Try structured content + if result.structuredContent: + return result.structuredContent + + return {"raw": str(result)} + + +async def read_resource(session: ClientSession, uri: str) -> str: + """Read a resource and return its content.""" + from pydantic import AnyUrl + + logger.info(f"Reading resource: {uri}") + + result = await session.read_resource(AnyUrl(uri)) + if result.contents and len(result.contents) > 0: + content = result.contents[0] + if isinstance(content, TextContent): + return content.text + return str(result) + + +async def run_happy_path(session: ClientSession): + """Run a complete shopping flow demonstrating UCP MCP capabilities.""" + + await print_separator("🛒 UCP MCP Client - Happy Path Demo") + print("This demo walks through a complete shopping experience using MCP.\n") + + # Step 1: Initialize connection + print("✅ Connected to UCP MCP Server") + await session.initialize() + + # Step 2: Display server capabilities + await display_tools(session) + await display_resources(session) + await display_prompts(session) + + # Step 3: Browse products + await print_separator("Step 1: Browse Products") + products = await call_tool(session, "list_products", {}) + + if "products" in products: + print(f"Found {products['total_count']} products:\n") + for p in products["products"]: + print(f" 🌸 {p['name']}") + print(f" ID: {p['id']}") + print(f" Price: ${p['price']:.2f}") + print(f" In Stock: {p['available_quantity']}") + print() + + # Step 4: Get details for a specific product + await print_separator("Step 2: Get Product Details") + product_id = "prod_roses_red" + product = await call_tool(session, "get_product", {"product_id": product_id}) + + if "error" not in product: + print(f"Product Details for {product['name']}:") + print(f" Description: {product['description']}") + print(f" Price: ${product['price']:.2f} {product.get('currency', 'USD')}") + print(f" Available: {product['available_quantity']} units") + print(f" Category: {product['category']}") + + # Step 5: Create checkout session + await print_separator("Step 3: Create Checkout Session") + checkout = await call_tool(session, "create_checkout", {}) + + if "error" not in checkout: + checkout_id = checkout["checkout_id"] + print(f"✅ Created checkout session: {checkout_id}") + print(f" Status: {checkout['status']}") + + # Step 6: Add items to cart + await print_separator("Step 4: Add Items to Cart") + + # Add roses + result = await call_tool(session, "add_to_checkout", { + "checkout_id": checkout_id, + "product_id": "prod_roses_red", + "quantity": 2 + }) + print(f"Added 2x Red Roses Bouquet") + print(f" Subtotal: ${result.get('subtotal', 0):.2f}") + + # Add sunflowers + result = await call_tool(session, "add_to_checkout", { + "checkout_id": checkout_id, + "product_id": "prod_sunflowers", + "quantity": 1 + }) + print(f"Added 1x Sunflower Arrangement") + print(f" Subtotal: ${result.get('subtotal', 0):.2f}") + print(f" Tax: ${result.get('tax', 0):.2f}") + print(f" Total: ${result.get('total', 0):.2f}") + + # Step 7: Set shipping address + await print_separator("Step 5: Set Shipping Address") + result = await call_tool(session, "set_shipping_address", { + "checkout_id": checkout_id, + "name": "Jane Doe", + "street": "123 Main Street", + "city": "San Francisco", + "state": "CA", + "postal_code": "94102", + "country": "US" + }) + print("✅ Shipping address set:") + addr = result.get("shipping_address", {}) + print(f" {addr.get('name')}") + print(f" {addr.get('street')}") + print(f" {addr.get('city')}, {addr.get('state')} {addr.get('postal_code')}") + print(f"\n Shipping Cost: ${result.get('shipping_cost', 0):.2f}") + print(f" Updated Total: ${result.get('total', 0):.2f}") + + # Step 8: Review checkout + await print_separator("Step 6: Review Checkout") + checkout_state = await call_tool(session, "get_checkout", { + "checkout_id": checkout_id + }) + + print("Current Checkout State:") + print(f" Status: {checkout_state.get('status')}") + print(f"\n Items:") + for item in checkout_state.get("line_items", []): + print(f" - {item['quantity']}x {item['name']}: ${item['total_price']:.2f}") + print(f"\n Subtotal: ${checkout_state.get('subtotal', 0):.2f}") + print(f" Tax: ${checkout_state.get('tax', 0):.2f}") + print(f" Shipping: ${checkout_state.get('shipping', 0):.2f}") + print(f" ──────────────") + print(f" TOTAL: ${checkout_state.get('total', 0):.2f} {checkout_state.get('currency', 'USD')}") + + # Step 9: Complete payment + await print_separator("Step 7: Complete Payment") + print("Processing payment...") + + order = await call_tool(session, "complete_payment", { + "checkout_id": checkout_id, + "payment_method": "mock_credit_card" + }) + + if "error" not in order: + print("\n🎉 Order Placed Successfully!") + print(f" Order ID: {order['order_id']}") + print(f" Status: {order['status']}") + print(f" Total: ${order['total']:.2f}") + print(f" Tracking: {order['tracking_number']}") + print(f"\n {order['message']}") + + # Step 10: Retrieve order + await print_separator("Step 8: Check Order Status") + order_details = await call_tool(session, "get_order", { + "order_id": order["order_id"] + }) + + print(f"Order {order_details['order_id']}:") + print(f" Status: {order_details['status']}") + print(f" Created: {order_details['created_at']}") + print(f" Payment: {order_details['payment_method']}") + print(f" Tracking: {order_details['tracking_number']}") + else: + print(f"❌ Payment failed: {order['error']}") + else: + print(f"❌ Failed to create checkout: {checkout['error']}") + + # Step 11: Read discovery profile resource + await print_separator("Bonus: Read Discovery Profile") + profile = await read_resource(session, "ucp://discovery/profile") + print("UCP Discovery Profile:") + print(json.dumps(json.loads(profile), indent=2)) + + await print_separator("🎉 Demo Complete!") + print("This demo showed the complete UCP shopping flow via MCP:") + print(" 1. Browsed product catalog") + print(" 2. Created checkout session") + print(" 3. Added items to cart") + print(" 4. Set shipping address") + print(" 5. Completed payment") + print(" 6. Retrieved order confirmation") + print("\nThe same flow can be used by AI agents to enable") + print("conversational commerce experiences!") + + +async def run_stdio_client(): + """Connect to the server using stdio transport.""" + server_params = StdioServerParameters( + command="uv", + args=["run", "ucp-mcp-server"], + ) + + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + await run_happy_path(session) + + +async def run_http_client(url: str): + """Connect to the server using HTTP transport.""" + from mcp.client.streamable_http import streamablehttp_client + + async with streamablehttp_client(url) as (read, write, _): + async with ClientSession(read, write) as session: + await run_happy_path(session) + + +def main(): + """Entry point for the UCP MCP Client.""" + parser = argparse.ArgumentParser( + description="UCP MCP Client - Demonstrate shopping via MCP" + ) + parser.add_argument( + "--transport", + choices=["stdio", "http"], + default="stdio", + help="Transport mechanism (default: stdio)", + ) + parser.add_argument( + "--url", + default="http://localhost:8000/mcp", + help="Server URL for HTTP transport (default: http://localhost:8000/mcp)", + ) + + args = parser.parse_args() + + if args.transport == "stdio": + asyncio.run(run_stdio_client()) + else: + asyncio.run(run_http_client(args.url)) + + +if __name__ == "__main__": + main() diff --git a/mcp/python/ucp_mcp_server.py b/mcp/python/ucp_mcp_server.py new file mode 100644 index 0000000..695d1fc --- /dev/null +++ b/mcp/python/ucp_mcp_server.py @@ -0,0 +1,980 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +UCP MCP Server - Model Context Protocol implementation for Universal Commerce Protocol. + +This module implements an MCP server that exposes UCP shopping capabilities as tools +and resources. It enables AI agents to interact with UCP-compliant merchants through +a standardized protocol, supporting operations like: + +- Product discovery and catalog browsing +- Checkout session management +- Payment processing +- Order tracking + +The server can operate in two modes: +1. Standalone mode with mock data for testing/development +2. Connected mode proxying to a real UCP REST server + +Usage: + # Run with stdio transport (for LLM integration): + uv run ucp-mcp-server + + # Run with streamable HTTP transport (for web clients): + uv run ucp-mcp-server --transport http --port 8000 +""" + +import argparse +import asyncio +import json +import logging +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Optional + +import httpx +from mcp.server.fastmcp import Context, FastMCP +from mcp.server.session import ServerSession +from pydantic import BaseModel + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Mock Data Models (for standalone testing) +# ============================================================================ + +@dataclass +class Product: + """Represents a product in the catalog.""" + id: str + name: str + description: str + price: float + currency: str = "USD" + available_quantity: int = 100 + category: str = "general" + image_url: Optional[str] = None + + +@dataclass +class LineItem: + """Represents an item in a checkout session.""" + product_id: str + name: str + quantity: int + unit_price: float + total_price: float + + +@dataclass +class CheckoutSession: + """Represents a checkout session.""" + id: str + status: str # "open", "pending_payment", "completed", "cancelled" + line_items: list[LineItem] = field(default_factory=list) + subtotal: float = 0.0 + tax: float = 0.0 + shipping: float = 0.0 + total: float = 0.0 + currency: str = "USD" + shipping_address: Optional[dict] = None + created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + + +@dataclass +class Order: + """Represents a completed order.""" + id: str + checkout_id: str + status: str # "confirmed", "processing", "shipped", "delivered", "cancelled" + line_items: list[LineItem] = field(default_factory=list) + total: float = 0.0 + currency: str = "USD" + shipping_address: Optional[dict] = None + payment_method: str = "mock_payment" + created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + tracking_number: Optional[str] = None + + +# ============================================================================ +# Mock Data Store +# ============================================================================ + +class MockStore: + """In-memory mock store for standalone testing.""" + + def __init__(self): + self.products: dict[str, Product] = {} + self.checkout_sessions: dict[str, CheckoutSession] = {} + self.orders: dict[str, Order] = {} + self._initialize_sample_products() + + def _initialize_sample_products(self): + """Initialize with sample flower shop products (matching REST sample).""" + sample_products = [ + Product( + id="prod_roses_red", + name="Red Roses Bouquet", + description="A beautiful bouquet of 12 fresh red roses", + price=49.99, + category="bouquets", + available_quantity=50, + ), + Product( + id="prod_tulips_mixed", + name="Mixed Tulips", + description="Colorful assortment of 15 spring tulips", + price=34.99, + category="bouquets", + available_quantity=30, + ), + Product( + id="prod_orchid_white", + name="White Orchid Plant", + description="Elegant white phalaenopsis orchid in ceramic pot", + price=79.99, + category="plants", + available_quantity=15, + ), + Product( + id="prod_sunflowers", + name="Sunflower Arrangement", + description="Bright sunflower arrangement with greenery", + price=44.99, + category="arrangements", + available_quantity=25, + ), + Product( + id="prod_lily_stargazer", + name="Stargazer Lilies", + description="Fragrant pink stargazer lily bouquet", + price=54.99, + category="bouquets", + available_quantity=20, + ), + ] + for product in sample_products: + self.products[product.id] = product + + +# Global store instance (for standalone mode) +mock_store = MockStore() + + +# ============================================================================ +# MCP Server Configuration +# ============================================================================ + +mcp = FastMCP( + name="UCP Shopping Service", + instructions=""" + This MCP server provides access to UCP (Universal Commerce Protocol) shopping + capabilities. You can: + + 1. Browse Products: Use list_products or get_product to explore the catalog + 2. Create Checkout: Use create_checkout to start a shopping session + 3. Add Items: Use add_to_checkout to add products to your cart + 4. Update Address: Use set_shipping_address to configure delivery + 5. Complete Purchase: Use complete_payment to finalize the order + 6. Track Orders: Use get_order to check order status + + Start by listing available products, then guide the user through checkout. + """, +) + + +# ============================================================================ +# Resources - Data Exposure (Read-only access to UCP data) +# ============================================================================ + +@mcp.resource("ucp://catalog/products") +def get_catalog() -> str: + """ + Get the complete product catalog. + + Returns a JSON array of all available products with their details + including name, description, price, and availability. + """ + products = [ + { + "id": p.id, + "name": p.name, + "description": p.description, + "price": p.price, + "currency": p.currency, + "available_quantity": p.available_quantity, + "category": p.category, + } + for p in mock_store.products.values() + ] + return json.dumps(products, indent=2) + + +@mcp.resource("ucp://catalog/products/{product_id}") +def get_product_resource(product_id: str) -> str: + """ + Get details for a specific product. + + Returns product information including name, description, price, + and current availability. + """ + product = mock_store.products.get(product_id) + if not product: + return json.dumps({"error": f"Product '{product_id}' not found"}) + + return json.dumps({ + "id": product.id, + "name": product.name, + "description": product.description, + "price": product.price, + "currency": product.currency, + "available_quantity": product.available_quantity, + "category": product.category, + }, indent=2) + + +@mcp.resource("ucp://checkout/{session_id}") +def get_checkout_resource(session_id: str) -> str: + """ + Get the current state of a checkout session. + + Returns the complete checkout state including items, totals, + and shipping information. + """ + session = mock_store.checkout_sessions.get(session_id) + if not session: + return json.dumps({"error": f"Checkout session '{session_id}' not found"}) + + return json.dumps({ + "id": session.id, + "status": session.status, + "line_items": [ + { + "product_id": item.product_id, + "name": item.name, + "quantity": item.quantity, + "unit_price": item.unit_price, + "total_price": item.total_price, + } + for item in session.line_items + ], + "subtotal": session.subtotal, + "tax": session.tax, + "shipping": session.shipping, + "total": session.total, + "currency": session.currency, + "shipping_address": session.shipping_address, + "created_at": session.created_at, + "updated_at": session.updated_at, + }, indent=2) + + +@mcp.resource("ucp://orders/{order_id}") +def get_order_resource(order_id: str) -> str: + """ + Get the status and details of an order. + + Returns complete order information including status, + items, and tracking details if available. + """ + order = mock_store.orders.get(order_id) + if not order: + return json.dumps({"error": f"Order '{order_id}' not found"}) + + return json.dumps({ + "id": order.id, + "checkout_id": order.checkout_id, + "status": order.status, + "line_items": [ + { + "product_id": item.product_id, + "name": item.name, + "quantity": item.quantity, + "unit_price": item.unit_price, + "total_price": item.total_price, + } + for item in order.line_items + ], + "total": order.total, + "currency": order.currency, + "shipping_address": order.shipping_address, + "payment_method": order.payment_method, + "created_at": order.created_at, + "tracking_number": order.tracking_number, + }, indent=2) + + +@mcp.resource("ucp://discovery/profile") +def get_discovery_profile() -> str: + """ + Get the UCP discovery profile for this merchant. + + Returns the merchant's UCP capabilities, supported services, + and available extensions following the UCP specification. + """ + return json.dumps({ + "ucp": { + "version": "2026-01-11", + "services": { + "dev.ucp.shopping": { + "version": "2026-01-11", + "spec": "https://ucp.dev/specs/shopping", + "mcp": { + "endpoint": "stdio://ucp-mcp-server", + }, + "rest": None, + "a2a": None, + } + }, + "capabilities": [ + { + "name": "dev.ucp.shopping.checkout", + "version": "2026-01-11", + "spec": "https://ucp.dev/specs/shopping/checkout", + }, + { + "name": "dev.ucp.shopping.fulfillment", + "version": "2026-01-11", + "spec": "https://ucp.dev/specs/shopping/fulfillment", + }, + ] + } + }, indent=2) + + +# ============================================================================ +# Tools - Actions (Operations with side effects) +# ============================================================================ + +@mcp.tool() +async def list_products( + category: str | None = None, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + List available products from the catalog. + + Args: + category: Optional filter by category (e.g., "bouquets", "plants", "arrangements") + + Returns: + List of products with their details and availability. + """ + if ctx: + await ctx.info("Fetching product catalog...") + + products = list(mock_store.products.values()) + + if category: + products = [p for p in products if p.category.lower() == category.lower()] + + return { + "products": [ + { + "id": p.id, + "name": p.name, + "description": p.description, + "price": p.price, + "currency": p.currency, + "available_quantity": p.available_quantity, + "category": p.category, + } + for p in products + ], + "total_count": len(products), + } + + +@mcp.tool() +async def get_product( + product_id: str, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Get detailed information about a specific product. + + Args: + product_id: The unique identifier of the product. + + Returns: + Product details including availability and pricing. + """ + product = mock_store.products.get(product_id) + + if not product: + return {"error": f"Product '{product_id}' not found"} + + if ctx: + await ctx.info(f"Found product: {product.name}") + + return { + "id": product.id, + "name": product.name, + "description": product.description, + "price": product.price, + "currency": product.currency, + "available_quantity": product.available_quantity, + "category": product.category, + "in_stock": product.available_quantity > 0, + } + + +@mcp.tool() +async def create_checkout( + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Create a new checkout session. + + Creates an empty checkout session that can be populated with items. + Use add_to_checkout to add products to the session. + + Returns: + The new checkout session with its unique ID. + """ + session_id = f"checkout_{uuid.uuid4().hex[:12]}" + session = CheckoutSession(id=session_id, status="open") + mock_store.checkout_sessions[session_id] = session + + if ctx: + await ctx.info(f"Created checkout session: {session_id}") + + return { + "checkout_id": session.id, + "status": session.status, + "message": "Checkout session created. Use add_to_checkout to add items.", + } + + +@mcp.tool() +async def add_to_checkout( + checkout_id: str, + product_id: str, + quantity: int = 1, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Add a product to an existing checkout session. + + Args: + checkout_id: The checkout session ID. + product_id: The product to add. + quantity: Number of items to add (default: 1). + + Returns: + Updated checkout state with current totals. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + if session.status != "open": + return {"error": f"Checkout session is {session.status}, cannot modify"} + + product = mock_store.products.get(product_id) + if not product: + return {"error": f"Product '{product_id}' not found"} + + if quantity > product.available_quantity: + return { + "error": f"Insufficient stock. Available: {product.available_quantity}" + } + + # Check if product already in cart, update quantity + existing_item = next( + (item for item in session.line_items if item.product_id == product_id), + None + ) + + if existing_item: + existing_item.quantity += quantity + existing_item.total_price = existing_item.quantity * existing_item.unit_price + else: + line_item = LineItem( + product_id=product.id, + name=product.name, + quantity=quantity, + unit_price=product.price, + total_price=product.price * quantity, + ) + session.line_items.append(line_item) + + # Recalculate totals + session.subtotal = sum(item.total_price for item in session.line_items) + session.tax = round(session.subtotal * 0.08, 2) # 8% tax + session.total = round(session.subtotal + session.tax + session.shipping, 2) + session.updated_at = datetime.utcnow().isoformat() + + if ctx: + await ctx.info(f"Added {quantity}x {product.name} to checkout") + + return { + "checkout_id": session.id, + "status": session.status, + "items_count": len(session.line_items), + "subtotal": session.subtotal, + "tax": session.tax, + "shipping": session.shipping, + "total": session.total, + "currency": session.currency, + } + + +@mcp.tool() +async def remove_from_checkout( + checkout_id: str, + product_id: str, + quantity: int | None = None, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Remove a product from the checkout session. + + Args: + checkout_id: The checkout session ID. + product_id: The product to remove. + quantity: Number of items to remove. If None, removes all. + + Returns: + Updated checkout state with current totals. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + if session.status != "open": + return {"error": f"Checkout session is {session.status}, cannot modify"} + + item_index = next( + (i for i, item in enumerate(session.line_items) if item.product_id == product_id), + None + ) + + if item_index is None: + return {"error": f"Product '{product_id}' not in checkout"} + + item = session.line_items[item_index] + + if quantity is None or quantity >= item.quantity: + session.line_items.pop(item_index) + removed_quantity = item.quantity + else: + item.quantity -= quantity + item.total_price = item.quantity * item.unit_price + removed_quantity = quantity + + # Recalculate totals + session.subtotal = sum(i.total_price for i in session.line_items) + session.tax = round(session.subtotal * 0.08, 2) + session.total = round(session.subtotal + session.tax + session.shipping, 2) + session.updated_at = datetime.utcnow().isoformat() + + if ctx: + await ctx.info(f"Removed {removed_quantity}x {item.name} from checkout") + + return { + "checkout_id": session.id, + "status": session.status, + "items_count": len(session.line_items), + "subtotal": session.subtotal, + "tax": session.tax, + "shipping": session.shipping, + "total": session.total, + "currency": session.currency, + } + + +@mcp.tool() +async def set_shipping_address( + checkout_id: str, + street: str, + city: str, + state: str, + postal_code: str, + country: str = "US", + name: str | None = None, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Set the shipping address for a checkout session. + + Args: + checkout_id: The checkout session ID. + street: Street address. + city: City name. + state: State/Province code. + postal_code: Postal/ZIP code. + country: Country code (default: "US"). + name: Recipient name (optional). + + Returns: + Updated checkout state with shipping calculated. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + if session.status not in ("open", "pending_payment"): + return {"error": f"Checkout session is {session.status}, cannot modify address"} + + session.shipping_address = { + "name": name, + "street": street, + "city": city, + "state": state, + "postal_code": postal_code, + "country": country, + } + + # Calculate shipping based on location (mock logic) + if country != "US": + session.shipping = 29.99 + elif state in ("AK", "HI"): + session.shipping = 19.99 + else: + session.shipping = 9.99 + + session.total = round(session.subtotal + session.tax + session.shipping, 2) + session.updated_at = datetime.utcnow().isoformat() + + if ctx: + await ctx.info(f"Set shipping address to {city}, {state}") + + return { + "checkout_id": session.id, + "status": session.status, + "shipping_address": session.shipping_address, + "shipping_cost": session.shipping, + "subtotal": session.subtotal, + "tax": session.tax, + "total": session.total, + "currency": session.currency, + } + + +@mcp.tool() +async def get_checkout( + checkout_id: str, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Get the current state of a checkout session. + + Args: + checkout_id: The checkout session ID. + + Returns: + Complete checkout state including items, totals, and shipping info. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + return { + "checkout_id": session.id, + "status": session.status, + "line_items": [ + { + "product_id": item.product_id, + "name": item.name, + "quantity": item.quantity, + "unit_price": item.unit_price, + "total_price": item.total_price, + } + for item in session.line_items + ], + "subtotal": session.subtotal, + "tax": session.tax, + "shipping": session.shipping, + "total": session.total, + "currency": session.currency, + "shipping_address": session.shipping_address, + "created_at": session.created_at, + "updated_at": session.updated_at, + } + + +@mcp.tool() +async def complete_payment( + checkout_id: str, + payment_method: str = "mock_credit_card", + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Complete payment and create an order. + + Args: + checkout_id: The checkout session ID. + payment_method: Payment method identifier (e.g., "mock_credit_card", "mock_paypal"). + + Returns: + Order confirmation with order ID and details. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + if session.status == "completed": + return {"error": "Checkout session already completed"} + + if not session.line_items: + return {"error": "Cannot complete payment with empty cart"} + + if not session.shipping_address: + return {"error": "Shipping address required before payment"} + + # Validate inventory + for item in session.line_items: + product = mock_store.products.get(item.product_id) + if not product or product.available_quantity < item.quantity: + return {"error": f"Insufficient stock for {item.name}"} + + if ctx: + await ctx.report_progress(0.3, 1.0, "Processing payment...") + + # Simulate payment processing (in real implementation, call payment provider) + await asyncio.sleep(0.5) # Simulate processing time + + if ctx: + await ctx.report_progress(0.6, 1.0, "Payment accepted, creating order...") + + # Create order + order_id = f"order_{uuid.uuid4().hex[:12]}" + order = Order( + id=order_id, + checkout_id=checkout_id, + status="confirmed", + line_items=session.line_items.copy(), + total=session.total, + currency=session.currency, + shipping_address=session.shipping_address, + payment_method=payment_method, + tracking_number=f"TRK{uuid.uuid4().hex[:10].upper()}", + ) + mock_store.orders[order_id] = order + + # Update inventory + for item in session.line_items: + product = mock_store.products.get(item.product_id) + if product: + product.available_quantity -= item.quantity + + # Update checkout status + session.status = "completed" + session.updated_at = datetime.utcnow().isoformat() + + if ctx: + await ctx.report_progress(1.0, 1.0, "Order created successfully!") + await ctx.info(f"Order {order_id} created successfully") + + return { + "order_id": order.id, + "checkout_id": checkout_id, + "status": order.status, + "total": order.total, + "currency": order.currency, + "tracking_number": order.tracking_number, + "shipping_address": order.shipping_address, + "items_count": len(order.line_items), + "message": "Order placed successfully! You will receive a confirmation email shortly.", + } + + +@mcp.tool() +async def get_order( + order_id: str, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Get the status and details of an order. + + Args: + order_id: The order ID. + + Returns: + Order details including status, items, and tracking info. + """ + order = mock_store.orders.get(order_id) + if not order: + return {"error": f"Order '{order_id}' not found"} + + if ctx: + await ctx.info(f"Retrieved order {order_id}: {order.status}") + + return { + "order_id": order.id, + "checkout_id": order.checkout_id, + "status": order.status, + "line_items": [ + { + "product_id": item.product_id, + "name": item.name, + "quantity": item.quantity, + "unit_price": item.unit_price, + "total_price": item.total_price, + } + for item in order.line_items + ], + "total": order.total, + "currency": order.currency, + "shipping_address": order.shipping_address, + "payment_method": order.payment_method, + "tracking_number": order.tracking_number, + "created_at": order.created_at, + } + + +@mcp.tool() +async def cancel_checkout( + checkout_id: str, + ctx: Context[ServerSession, None] | None = None +) -> dict: + """ + Cancel an open checkout session. + + Args: + checkout_id: The checkout session ID to cancel. + + Returns: + Confirmation of cancellation. + """ + session = mock_store.checkout_sessions.get(checkout_id) + if not session: + return {"error": f"Checkout session '{checkout_id}' not found"} + + if session.status == "completed": + return {"error": "Cannot cancel a completed checkout"} + + if session.status == "cancelled": + return {"error": "Checkout session already cancelled"} + + session.status = "cancelled" + session.updated_at = datetime.utcnow().isoformat() + + if ctx: + await ctx.info(f"Cancelled checkout session {checkout_id}") + + return { + "checkout_id": session.id, + "status": session.status, + "message": "Checkout session cancelled successfully", + } + + +# ============================================================================ +# Prompts - Reusable conversation patterns +# ============================================================================ + +@mcp.prompt(title="Shopping Assistant Introduction") +def shopping_intro() -> str: + """Generate an introduction for the shopping assistant.""" + return """You are a helpful shopping assistant for a flower shop. +You can help customers: +1. Browse our product catalog (use list_products) +2. Get details about specific products (use get_product) +3. Create a checkout session (use create_checkout) +4. Add items to their cart (use add_to_checkout) +5. Set shipping address (use set_shipping_address) +6. Complete their purchase (use complete_payment) +7. Track their orders (use get_order) + +Start by asking what the customer is looking for today.""" + + +@mcp.prompt(title="Order Confirmation") +def order_confirmation(order_id: str, total: str) -> str: + """Generate an order confirmation message.""" + return f"""The customer's order has been placed successfully! + +Order ID: {order_id} +Total: {total} + +Please confirm the order details with the customer and let them know: +1. They will receive a confirmation email shortly +2. Their order will be processed within 1-2 business days +3. They can track their order using the order ID + +Ask if there's anything else you can help them with.""" + + +@mcp.prompt(title="Product Recommendation") +def recommend_products(occasion: str) -> str: + """Generate product recommendations for an occasion.""" + return f"""The customer is looking for flowers for: {occasion} + +Based on this occasion, use the list_products tool to find suitable options, +then recommend 2-3 products that would be perfect for this occasion. + +Consider: +- Price range appropriate for the occasion +- Flower meanings and symbolism +- Seasonal availability + +Provide personalized recommendations with reasons why each would be a good choice.""" + + +# ============================================================================ +# Entry Point +# ============================================================================ + +def main(): + """Entry point for the UCP MCP Server.""" + parser = argparse.ArgumentParser( + description="UCP MCP Server - Model Context Protocol for Universal Commerce" + ) + parser.add_argument( + "--transport", + choices=["stdio", "http", "sse"], + default="stdio", + help="Transport mechanism (default: stdio)", + ) + parser.add_argument( + "--port", + type=int, + default=8000, + help="Port for HTTP/SSE transport (default: 8000)", + ) + parser.add_argument( + "--host", + default="0.0.0.0", + help="Host for HTTP/SSE transport (default: 0.0.0.0)", + ) + + args = parser.parse_args() + + logger.info(f"Starting UCP MCP Server with {args.transport} transport") + + if args.transport == "stdio": + mcp.run(transport="stdio") + elif args.transport == "http": + mcp.settings.host = args.host + mcp.settings.port = args.port + mcp.run(transport="streamable-http") + elif args.transport == "sse": + mcp.settings.host = args.host + mcp.settings.port = args.port + mcp.run(transport="sse") + + +if __name__ == "__main__": + main()