DEV Community

Cover image for Building Your First MCP Server
Chandra Shettigar
Chandra Shettigar

Posted on • Edited on

Building Your First MCP Server

#genai #mcp #aiagent

A hands-on tutorial to create an MCP-enabled product database service

In our previous posts, we've seen what MCP is and why it matters. Now let's get our hands dirty and build an actual MCP server that any AI agent can use.

We'll create a Product Database MCP Server that exposes product information through standardized MCP tools. By the end, you'll have a working MCP server and understand exactly how to make any service AI-agent ready.

🚀 Want to jump straight to the code? Check out the complete implementation on GitHub: mcp-product-search-service

Current MCP Ecosystem Reality

What works today:

  • MCP protocol specification is stable
  • You can build MCP servers (like we're about to do)
  • HTTP-based integration works universally

What's still developing:

  • Official Python MCP libraries may have different APIs than shown
  • AI agent framework MCP support varies by platform
  • Ecosystem tooling is rapidly emerging

Bottom line: Focus on the HTTP API patterns shown here - they're universal and future-proof regardless of how MCP libraries evolve.

What We're Building

A product database service that provides:

  • search_products: Find products by name or category
  • get_product_details: Get detailed information about a specific product
  • check_inventory: Check stock levels for a product
  • REST API endpoints: Direct HTTP access alongside MCP protocol
  • Comprehensive testing: Full test suite with 18+ tests
  • Docker support: Containerized development environment

Prerequisites

We'll use a modern Python stack with FastAPI and handle MCP protocol manually:

# Clone the complete project (when available)
git clone [email protected]:devteds/mcp-test-service.git
cd mcp-product-search-service

# Or start from scratch with these dependencies
pip install fastapi uvicorn pydantic pytest httpx
Enter fullscreen mode Exit fullscreen mode

Note: We're implementing the MCP protocol directly rather than relying on potentially unstable MCP libraries.

Step 1: Project Structure

Unlike simple single-file examples, we'll build a proper service:

mcp-product-search-service/
├── mcp_service/
│   ├── __init__.py
│   ├── __main__.py          # Entry point for python -m mcp_service
│   ├── main.py              # Application startup
│   ├── server.py            # FastAPI app configuration
│   ├── handlers.py          # MCP protocol handlers
│   ├── models.py            # Pydantic data models
│   └── data.py              # Product database
├── tests/
│   ├── test_server.py       # Basic server tests
│   └── test_product_service.py  # MCP functionality tests
├── requirements.txt
├── pyproject.toml
└── README.md
Enter fullscreen mode Exit fullscreen mode

Step 2: The Product Database

First, let's set up our product data in mcp_service/data.py:

"""Product database for the MCP service."""

# Our sample product database
PRODUCTS = [
    {
        "id": "1",
        "name": "iPhone 15 Pro",
        "category": "Electronics",
        "price": 999.99,
        "stock": 50,
        "description": "Latest iPhone with titanium design"
    },
    {
        "id": "2", 
        "name": "MacBook Air M3",
        "category": "Electronics", 
        "price": 1299.99,
        "stock": 25,
        "description": "Lightweight laptop with M3 chip"
    },
    {
        "id": "3",
        "name": "Nike Air Max",
        "category": "Footwear",
        "price": 129.99, 
        "stock": 100,
        "description": "Classic running shoes"
    },
    {
        "id": "4",
        "name": "Coffee Maker Pro",
        "category": "Appliances",
        "price": 199.99,
        "stock": 15,
        "description": "Professional grade coffee maker"
    }
]

def search_products(query: str = "", category: str = "") -> list:
    """Search for products by name or category"""
    results = []

    for product in PRODUCTS:
        name_match = query.lower() in product["name"].lower() if query else True
        category_match = category.lower() == product["category"].lower() if category else True

        if name_match and category_match:
            results.append({
                "id": product["id"],
                "name": product["name"], 
                "category": product["category"],
                "price": product["price"]
            })

    return results

def get_product_details(product_id: str) -> dict:
    """Get detailed information about a specific product"""
    for product in PRODUCTS:
        if product["id"] == product_id:
            return product
    return {"error": "Product not found"}

def check_inventory(product_id: str) -> dict:
    """Check stock levels for a product"""
    for product in PRODUCTS:
        if product["id"] == product_id:
            return {
                "product_id": product_id,
                "product_name": product["name"],
                "stock": product["stock"],
                "in_stock": product["stock"] > 0
            }
    return {"error": "Product not found"}
Enter fullscreen mode Exit fullscreen mode

Step 3: MCP Protocol Implementation

The heart of our MCP server is in mcp_service/handlers.py. Here's how we handle MCP messages:

from fastapi import APIRouter
from pydantic import BaseModel
from typing import Dict, Any, Optional
from .data import search_products, get_product_details, check_inventory

router = APIRouter()

class MCPRequest(BaseModel):
    id: str
    method: str
    params: Dict[str, Any] = {}

class MCPResponse(BaseModel):
    id: str
    result: Optional[Dict[str, Any]] = None
    error: Optional[Dict[str, Any]] = None

@router.post("/mcp/message", response_model=MCPResponse)
async def handle_mcp_message(request: MCPRequest) -> MCPResponse:
    """Handle incoming MCP messages for product operations."""
    try:
        if request.method == "capabilities":
            return MCPResponse(
                id=request.id,
                result={
                    "capabilities": {
                        "tools": [
                            {
                                "name": "search_products",
                                "description": "Search for products by name or category",
                                "parameters": {
                                    "query": {"type": "string", "description": "Search query"},
                                    "category": {"type": "string", "description": "Product category"}
                                }
                            },
                            {
                                "name": "get_product_details", 
                                "description": "Get detailed information about a specific product",
                                "parameters": {
                                    "product_id": {"type": "string", "required": True}
                                }
                            },
                            {
                                "name": "check_inventory",
                                "description": "Check stock levels for a product",
                                "parameters": {
                                    "product_id": {"type": "string", "required": True}
                                }
                            }
                        ]
                    }
                }
            )

        elif request.method == "search_products":
            query = request.params.get("query", "")
            category = request.params.get("category", "")
            results = search_products(query, category)
            return MCPResponse(
                id=request.id,
                result={"products": results, "count": len(results)}
            )

        elif request.method == "get_product_details":
            product_id = request.params.get("product_id")
            if not product_id:
                return MCPResponse(
                    id=request.id,
                    error={"code": -32602, "message": "Missing required parameter: product_id"}
                )

            product = get_product_details(product_id)
            return MCPResponse(id=request.id, result=product)

        elif request.method == "check_inventory":
            product_id = request.params.get("product_id")
            if not product_id:
                return MCPResponse(
                    id=request.id,
                    error={"code": -32602, "message": "Missing required parameter: product_id"}
                )

            inventory = check_inventory(product_id)
            return MCPResponse(id=request.id, result=inventory)

        else:
            return MCPResponse(
                id=request.id,
                error={"code": -32601, "message": f"Method '{request.method}' not found"}
            )

    except Exception as e:
        return MCPResponse(
            id=request.id,
            error={"code": -32603, "message": f"Internal error: {str(e)}"}
        )
Enter fullscreen mode Exit fullscreen mode

Step 4: Start Your MCP Server

Our service includes a proper entry point in mcp_service/main.py:

"""Main entry point for the MCP service."""

import uvicorn

def main():
    """Main entry point for the MCP service."""
    uvicorn.run(
        "mcp_service.server:create_app",
        factory=True,
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

Start the server:

python -m mcp_service
Enter fullscreen mode Exit fullscreen mode

You should see:

INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [7014] using WatchFiles
INFO:     Started server process [7017]
INFO:     Application startup complete.
Enter fullscreen mode Exit fullscreen mode

Step 5: Test Your MCP Server

Test MCP Protocol Directly 

# Test capabilities discovery
curl -X POST http://localhost:8000/api/v1/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"id": "1", "method": "capabilities", "params": {}}'

# Search for products
curl -X POST http://localhost:8000/api/v1/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"id": "2", "method": "search_products", "params": {"query": "iPhone"}}'

# Get product details
curl -X POST http://localhost:8000/api/v1/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"id": "3", "method": "get_product_details", "params": {"product_id": "1"}}'
Enter fullscreen mode Exit fullscreen mode

Test REST API Endpoints

Our service also provides REST endpoints for direct access:

# Search products via REST
curl "http://localhost:8000/api/v1/products/search?query=MacBook"

# Get product details via REST  
curl "http://localhost:8000/api/v1/products/2"

# Check inventory via REST
curl "http://localhost:8000/api/v1/products/3/inventory"

# Get available categories
curl "http://localhost:8000/api/v1/categories"
Enter fullscreen mode Exit fullscreen mode

Step 6: Automated Testing

Our service includes comprehensive tests. Run them with:

pytest tests/ -v
Enter fullscreen mode Exit fullscreen mode

Sample test output:

tests/test_product_service.py::TestMCPEndpoints::test_search_products_message PASSED
tests/test_product_service.py::TestMCPEndpoints::test_get_product_details_message PASSED
tests/test_product_service.py::TestMCPEndpoints::test_check_inventory_message PASSED
tests/test_server.py::test_root_endpoint PASSED

============================================ 18 passed in 0.40s ============================================
Enter fullscreen mode Exit fullscreen mode

Step 7: Connect to an AI Agent

Here's how an AI agent would discover and use your MCP server. We'll show multiple approaches since MCP client libraries are still evolving:

Using Direct HTTP (Recommended - Universal) 

import asyncio
import httpx

class MCPProductClient:
    """A client for interacting with our Product Search MCP server."""

    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url

    async def discover_capabilities(self):
        """Discover what tools the MCP server provides."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/mcp/message",
                json={
                    "id": "capabilities",
                    "method": "capabilities",
                    "params": {}
                }
            )
            return response.json()

    async def call_tool(self, tool_name: str, arguments: dict):
        """Call a tool on the MCP server."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/mcp/message",
                json={
                    "id": f"call_{tool_name}",
                    "method": tool_name,
                    "params": arguments
                }
            )
            return response.json()

    async def search_products(self, query: str = "", category: str = ""):
        """Search for products."""
        result = await self.call_tool("search_products", {
            "query": query,
            "category": category
        })
        return result.get("result", {})

# Usage example
async def demo():
    client = MCPProductClient("http://localhost:8000")

    # Discover available tools
    capabilities = await client.discover_capabilities()
    tools = capabilities["result"]["capabilities"]["tools"]
    print("Available tools:", [tool["name"] for tool in tools])

    # Search for electronics
    electronics = await client.search_products(category="Electronics")
    print(f"Found {electronics['count']} electronics")

    # Get product details
    product_result = await client.call_tool("get_product_details", {"product_id": "1"})
    iphone = product_result["result"]
    print(f"iPhone: {iphone['name']} - ${iphone['price']}")

# Run the demo
asyncio.run(demo())
Enter fullscreen mode Exit fullscreen mode

Using Hypothetical MCP Library (Future) 

# Note: This example uses hypothetical MCP client patterns
# Actual implementation may vary based on MCP library evolution

try:
    from mcp import types  # May not exist yet

    class TypedMCPClient:
        """Example using potential future MCP types for better type safety."""

        def __init__(self, base_url: str = "http://localhost:8000"):
            self.base_url = base_url

        async def call_tool_typed(self, tool_name: str, arguments: dict):
            """Call a tool using potential future MCP types."""
            # This is conceptual - actual MCP library API may differ
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f"{self.base_url}/api/v1/mcp/message",
                    json={
                        "id": "typed_call",
                        "method": tool_name,
                        "params": arguments or {}
                    }
                )
                return response.json()

except ImportError:
    print("MCP library not available - use direct HTTP approach above")
Enter fullscreen mode Exit fullscreen mode

Integration with AI Frameworks 

# Example of how this might integrate with LangChain (conceptual)
class MCPTool:
    """Wrapper to use our MCP server as a LangChain tool."""

    def __init__(self, mcp_client: MCPProductClient):
        self.client = mcp_client

    async def search_products(self, query: str, category: str = ""):
        """LangChain-compatible tool for product search."""
        result = await self.client.search_products(query, category)
        return f"Found {result['count']} products: {result['products']}"

    async def get_product(self, product_id: str):
        """LangChain-compatible tool for product details."""
        result = await self.client.call_tool("get_product_details", {"product_id": product_id})
        return result["result"]
Enter fullscreen mode Exit fullscreen mode

What Just Happened?

You built a MCP ecosystem!

  1. Proper Project Structure: Modular, testable, maintainable code
  2. MCP Protocol: Full JSON-RPC 2.0 implementation with error handling
  3. Tool Discovery: AI agents can discover capabilities automatically
  4. Dual Interface: Both MCP protocol and REST API access
  5. Comprehensive Testing: 18+ tests covering all functionality

The Power of What You Built

Your MCP server can now be used by:

  • Any HTTP client via direct API calls (works today)
  • Future MCP-compatible clients using the MCP protocol
  • AI agent frameworks with MCP support or custom HTTP adapters
  • Applications that implement MCP client functionality
  • Direct integration in any system that can make HTTP requests

Reality Check: While MCP is designed to work with any AI agent framework, most frameworks are still adding MCP support. However, your server provides both MCP protocol and REST API access, making it universally accessible.

Key Features of Our Implementation

🔍 Discovery Endpoint

The capabilities method tells AI agents exactly what your service can do:

{
  "capabilities": {
    "tools": [
      {"name": "search_products", "description": "Search for products by name or category"},
      {"name": "get_product_details", "description": "Get detailed information about a specific product"},
      {"name": "check_inventory", "description": "Check stock levels for a product"}
    ]
  }
}
Enter fullscreen mode Exit fullscreen mode

🛡️ Error Handling

Proper MCP error codes and messages:

  • -32601: Method not found
  • -32602: Invalid parameters
  • -32603: Internal error

📊 Structured Responses

All responses follow MCP protocol with proper id, result, and error fields.

Adding More Functionality

Want to add a new tool? Just add a method to your handlers:

elif request.method == "update_inventory":
    product_id = request.params.get("product_id")
    new_stock = request.params.get("new_stock")

    if not product_id or new_stock is None:
        return MCPResponse(
            id=request.id,
            error={"code": -32602, "message": "Missing required parameters"}
        )

    # Update logic here (in real app, update database)
    result = {"product_id": product_id, "new_stock": new_stock, "updated": True}

    return MCPResponse(id=request.id, result=result)
Enter fullscreen mode Exit fullscreen mode

That's it! Any AI agent using your MCP server will automatically discover and can use the new tool.

Real-World Next Steps

To make this production-ready:

  1. Replace sample data with real database (PostgreSQL, MongoDB, etc.)
  2. Add authentication for secure access (API keys, JWT tokens)
  3. Implement rate limiting to prevent abuse
  4. Add comprehensive logging for monitoring and debugging
  5. Deploy to cloud (AWS, GCP, Azure, etc.)
  6. Add more tools (create, update, delete products)
  7. Add data validation and sanitization
  8. Implement caching for better performance

Complete Implementation

📚 Repository Note: The complete implementation is here - https://github.com/devteds/mcp-test-service. The patterns shown here are tested and work, but check the latest MCP documentation for any protocol updates.

The implementation would include:

  • ✅ Complete MCP server implementation
  • ✅ Comprehensive test suite (18+ tests)
  • ✅ Docker development environment
  • ✅ API documentation
  • ✅ Usage examples

Key Takeaways

  • MCP servers are powerful - standardized protocol for AI integration
  • HTTP-first approach ensures universal compatibility
  • Tool discovery makes your services self-documenting for AI agents
  • Production patterns matter - proper structure, testing, error handling
  • Focus on business logic, not integration complexity
  • MCP libraries are evolving - HTTP APIs provide stability

🧪 Testing Note: The code shown has been tested locally but MCP libraries evolve rapidly. The HTTP-based approach provides maximum compatibility regardless of library changes.

⚠️ Important Note: The AI space evolves incredibly fast. MCP is still in early stages and may evolve significantly - what we know today might look different in a few months. Also note that this example MCP server hasn't been tested by integrating into any MCP host yet.

The key is building your foundation in AI concepts and staying adaptable as the ecosystem matures. Keep learning, keep experimenting!

📝 Code Note: The code examples use direct HTTP implementation of MCP protocol for maximum compatibility. Official MCP libraries may have different APIs - always refer to current MCP documentation for the latest patterns.

Next up: We'll explore the full MCP ecosystem and see real-world examples of MCP servers in production.

What You Built Today

✅ MCP server with proper structure

✅ Tool discovery capability for AI agents

✅ Comprehensive testing and error handling

✅ Universal HTTP compatibility regardless of MCP library evolution

✅ Foundation for real-world deployment

Ready to turn your services into MCP servers? Clone this repository!

Top comments (0)