MCP Tooling bauen

Übersicht

Mit dem Model Context Protocol (MCP) können Sie eigene Tools und Datenquellen für meinGPT erstellen. Diese Anleitung zeigt Ihnen, wie Sie mit FastMCP schnell und effizient MCP-Server entwickeln und über den HTTP Streamable Transport in meinGPT einbinden.

Was ist MCP?

Das Model Context Protocol (MCP) ist ein standardisiertes Protokoll für die Kommunikation zwischen LLMs und externen Tools. MCP-Server können:

Tools bereitstellen - Funktionen, die das LLM ausführen kann
Resources anbieten - Datenquellen, die das LLM lesen kann
Prompts definieren - Wiederverwendbare Vorlagen für Interaktionen

FastMCP Installation

pip install fastmcp uvicorn

Minimales Beispiel

Erstellen Sie einen einfachen MCP-Server mit einem Tool:

from fastmcp import FastMCP

mcp = FastMCP("My MCP Server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Start with: uvicorn main:mcp --port 8000

FastAPI Integration

FastMCP lässt sich nahtlos in FastAPI-Anwendungen integrieren:

from fastmcp import FastMCP
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# Create FastAPI app
app = FastAPI()

# Add CORS middleware for browser clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Create MCP server
mcp = FastMCP("My MCP Server")

@mcp.tool()
async def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    return a * b

# Mount MCP as ASGI app
mcp_app = mcp.http_app(path='/mcp')
app.mount("/", mcp_app)

# Add health check
@app.get("/health")
async def health():
    return {"status": "healthy"}

# Start: uvicorn main:app --reload
# MCP URL: http://localhost:8000/mcp

Authentication

FastMCP bietet flexible Authentifizierungsoptionen:

from fastmcp import FastMCP
from fastmcp.server.dependencies import get_http_request
from fastapi import HTTPException, status

mcp = FastMCP("Protected Server")

# API key management
API_KEYS = {"secret-key-1": "Production API Key"}

@mcp.tool()
async def protected_function(data: str) -> dict:
    """Protected function requiring authentication"""
    request = get_http_request()
    
    # Check API key from header
    api_key = request.headers.get("X-API-Key")
    
    if api_key not in API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )
    
    return {"status": "success", "data": data}

Resources

Resources bieten strukturierte Daten für das LLM:

@mcp.resource("config://settings")
async def get_settings():
    """Get current application settings"""
    return {
        "version": "1.0.0",
        "environment": "production",
        "features": ["api", "auth", "logging"]
    }

@mcp.resource("data://{category}/{id}")
async def get_data(category: str, id: str):
    """Get data by category and ID"""
    # Fetch from database
    data = await fetch_from_db(category, id)
    return data

Error Handling

Robuste Fehlerbehandlung ist essentiell:

import httpx
from typing import Optional, Dict, Any

@mcp.tool()
async def api_call(
    endpoint: str, 
    method: str = "GET",
    data: Optional[Dict[str, Any]] = None
) -> dict:
    """Make HTTP request with error handling"""
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.request(
                method=method,
                url=endpoint,
                json=data
            )
            response.raise_for_status()
            
            return {
                "success": True,
                "data": response.json(),
                "status_code": response.status_code
            }
            
        except httpx.HTTPStatusError as e:
            return {
                "success": False,
                "error": f"HTTP {e.response.status_code}",
                "status_code": e.response.status_code
            }
        except httpx.TimeoutException:
            return {
                "success": False,
                "error": "Request timeout"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

Context und Progress

Nutzen Sie Context für detailliertes Logging:

from fastmcp import Context

@mcp.tool()
async def process_data(file_path: str, ctx: Context) -> dict:
    """Process file with progress reporting"""
    # Log information
    await ctx.info(f"Processing file: {file_path}")
    
    # Read file
    data = await read_file(file_path)
    
    # Report progress
    await ctx.report_progress(50, f"Processing {len(data)} items")
    
    # Process data
    results = await process_items(data)
    
    await ctx.report_progress(100, "Processing complete")
    
    return {"processed": len(results), "results": results}

Best Practices

1. Klare Tool-Dokumentation

@mcp.tool()
async def send_email(to: str, subject: str, body: str) -> dict:
    """
    Send an email notification.
    
    Use this when:
    - User explicitly requests to send an email
    - System needs to send notifications
    - Alerts need to be triggered
    
    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email content (plain text)
    
    Returns:
        Dict with status and message_id
    """
    # Implementation

2. Strukturierte Responses

from pydantic import BaseModel
from datetime import datetime

class APIResponse(BaseModel):
    success: bool
    data: Any = None
    error: str = None
    timestamp: datetime
    
@mcp.tool()
async def fetch_data(query: str) -> APIResponse:
    """Fetch data with structured response"""
    try:
        result = await database.query(query)
        return APIResponse(
            success=True,
            data=result,
            timestamp=datetime.now()
        )
    except Exception as e:
        return APIResponse(
            success=False,
            error=str(e),
            timestamp=datetime.now()
        )

3. Dependency Injection

from fastmcp.server.dependencies import get_http_request
from typing import Annotated
from fastapi import Header

@mcp.tool()
async def database_query(
    query: str,
    db_url: Annotated[str, Header(alias="X-Database-URL")]
) -> list:
    """Execute database query with injected connection"""
    # Use db_url from header
    conn = await get_connection(db_url)
    return await conn.fetch(query)

In meinGPT einbinden

Starten Sie Ihren MCP-Server
Öffnen Sie die meinGPT-Einstellungen
Navigieren Sie zu "MCP Server"
Fügen Sie einen neuen Server hinzu:
- Name: Ihr Server-Name
- URL: http://localhost:8000/mcp
- Transport: HTTP Streamable
- Optional: Header für Authentifizierung

Deployment

Für Produktion empfiehlt sich Containerisierung:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Weiterführende Ressourcen

FastMCP Dokumentation - Vollständige Referenz
MCP Spezifikation - Protokoll Details
FastMCP GitHub - Beispiele und Code

Beispielprojekte

Weitere Beispiele für MCP-Server finden Sie in der FastMCP Beispielsammlung.