MCP Server Implementation Pattern

Minimal Document Search MCP Server (Python)

"""
Minimal MCP server exposing a document search tool.
This pattern shows the essential structure - production code needs error handling.
"""
import asyncio
import json
from typing import Any
from mcp import Server, Tool, Resource
from mcp.types import TextContent

# Initialize MCP server {.unnumbered}
server = Server("document-search")

# Define available tools {.unnumbered}
@server.tool("search_documents")
async def search_documents(query: str, top_k: int = 5) -> list[dict]:
    """
    Search internal document corpus.
    
    Args:
        query: Natural language search query
        top_k: Number of results to return (default: 5, max: 20)
    
    Returns:
        List of matching documents with relevance scores
    """
    # In production: call your vector DB here
    results = await vector_store.search(
        query=query,
        limit=min(top_k, 20),
        include_metadata=True
    )
    
    return [
        {
            "doc_id": r.id,
            "title": r.metadata.get("title", "Untitled"),
            "snippet": r.text[:500],
            "score": r.score,
            "source": r.metadata.get("source_url")
        }
        for r in results
    ]

@server.tool("get_document")
async def get_document(doc_id: str) -> dict:
    """
    Retrieve full document content by ID.
    
    Args:
        doc_id: Unique document identifier from search results
    
    Returns:
        Full document content and metadata
    """
    doc = await document_store.get(doc_id)
    if not doc:
        return {"error": f"Document {doc_id} not found"}
    
    return {
        "doc_id": doc_id,
        "title": doc.title,
        "content": doc.content,
        "metadata": doc.metadata,
        "last_updated": doc.updated_at.isoformat()
    }

# Expose corpus statistics as a resource {.unnumbered}
@server.resource("corpus://stats")
async def corpus_stats() -> Resource:
    """Current corpus statistics - refreshed hourly."""
    stats = await get_corpus_stats()
    return Resource(
        uri="corpus://stats",
        name="Corpus Statistics",
        description="Document corpus size and freshness metrics",
        content=TextContent(
            text=json.dumps({
                "total_documents": stats.doc_count,
                "total_chunks": stats.chunk_count,
                "last_indexed": stats.last_index_time.isoformat(),
                "index_health": stats.health_status
            }, indent=2)
        )
    )

# Server lifecycle {.unnumbered}
async def main():
    async with server.run_stdio() as running:
        await running.wait_closed()

if __name__ == "__main__":
    asyncio.run(main())

MCP Server Manifest (mcp.json)

{
  "name": "document-search",
  "version": "1.0.0",
  "description": "Search and retrieve internal documentation",
  "tools": [
    {
      "name": "search_documents",
      "description": "Semantic search across document corpus",
      "parameters": {
        "type": "object",
        "properties": {
          "query": {
            "type": "string",
            "description": "Natural language search query"
          },
          "top_k": {
            "type": "integer",
            "default": 5,
            "maximum": 20
          }
        },
        "required": ["query"]
      }
    },
    {
      "name": "get_document",
      "description": "Retrieve full document by ID",
      "parameters": {
        "type": "object",
        "properties": {
          "doc_id": {
            "type": "string",
            "description": "Document ID from search results"
          }
        },
        "required": ["doc_id"]
      }
    }
  ],
  "resources": [
    {
      "uri": "corpus://stats",
      "name": "Corpus Statistics",
      "description": "Current index size and health"
    }
  ]
}

Key Implementation Notes

• Transport: Use stdio for local development, HTTP+SSE for production
• Security: Scope capabilities carefully, validate all inputs
• Lifecycle: Handle graceful shutdown and reconnection
• Error handling: Return structured errors, don’t crash the server