MCP Tooling bauen

Erstelle eigene MCP-Server mit FastMCP und integriere diese in meinGPT

Übersicht

Mit dem Model Context Protocol (MCP) kannst Du eigene Tools und Datenquellen für meinGPT erstellen. Diese Anleitung zeigt Dir, wie Du mit FastMCP schnell und effizient MCP-Server entwickelst und über den HTTP Streamable Transport in meinGPT einbindest.

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

Erstelle 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

Nutze 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

Starte Deinen MCP-Server
Füge ihn als Custom MCP Server hinzu, entweder unter Einstellungen → Verbindungen → Eigene MCP-Server → MCP-Server hinzufügen, oder direkt im Assistenten-Editor über den Tab Tools → Custom MCP
Fülle das Formular aus:
- Name: Dein Server-Name
- URL: http://localhost:8000/mcp
- Optional: Header für Authentifizierung
- Optional: Autorisierung (OAuth) oder Signierte Nutzeridentität, falls Dein Server das braucht

Details zu den Feldern und Authentifizierungsoptionen findest Du unter Custom MCP Server.

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 findest Du in der FastMCP Beispielsammlung.

War diese Seite hilfreich?