MCP-Server bauen, die ki-web-Anleitung

Diese Seite zeigt Schritt für Schritt, wie der MCP-Server ki-web entstanden ist: ein stdio-MCP in Python, der die Verwaltung der Seiten auf ki.karlkratz.com an das PHP-Backend unter /api/pages delegiert. Diese Anleitung beschreibt jeden Schritt, sodass vergleichbare Adapter mit überschaubarem Aufwand nachgebaut werden können.

Die hier verwendeten Konventionen sind formal hinterlegt in /mcp-server-erstellen-regeln und maschinenlesbar unter /var/www/code/python-mcp/.

1. Was am Ende läuft

Fünf Tools, alle über stdio erreichbar für Claude Code, qwen-code und Codex:

Tool	Wirkung
list_pages	Suche, Sortierung, Pagination
get_page	Eine Seite per Slug
create_page	Neue Seite anlegen
update_page	Partial-Update, nur geänderte Felder
delete_page	Seite per Slug löschen

Jeder Tool-Aufruf landet als HTTP-Request bei /api/pages; Slug-Regex, Length-Limits und HTML-Sanitization passieren ausschließlich im PHP-Backend. Der Python-Code ist nur ein Adapter, die fachlichen Regeln liegen damit nur einmal vor.

2. Voraussetzungen

Python 3.13 mit venv unter /opt/venv/main
Pakete mcp >= 1.27, httpx >= 0.28, pydantic >= 2.13 (vor-installiert)
Backend mit JSON-REST-Schnittstelle, hier die PHP-App unter /api/pages
Schreibrecht auf /var/www/mcp und /usr/local/bin

3. Verzeichnisstruktur

/var/www/mcp/
  pyproject.toml
  requirements.lock
  README.md
  ki_web_mcp/
    __init__.py
    server.py             # FastMCP + Tool-Registrierung
    composition.py        # Composition-Root
    config.py             # Frozen Config + Env-Loader
    schemas.py            # Literal-Enums, Annotated-Bounds, TypedDicts
    domain/
      __init__.py
      page.py             # Page-Dataclass
      errors.py           # BackendError, TransportError
    application/
      __init__.py
      ports.py            # PagesPort als typing.Protocol
      use_cases.py        # PageUseCases, orchestriert den Port
    infrastructure/
      __init__.py
      pages_client.py     # httpx-Adapter, implementiert PagesPort
      page_mapper.py      # JSON <-> Page-Mapping
  tests/
    __init__.py
    conftest.py           # FakePages-Port + Fixtures
    test_use_cases.py     # Use-Case-Unit-Tests
    test_schema.py        # tools/list-Drift via Subprocess
    test_e2e.py           # Lifecycle gegen Live-Backend

4. pyproject.toml

Echtes Python-Package, keine sys.path-Hacks. Entry-Point heißt ki-web, der Wrapper unter /usr/local/bin/ki-web ruft ihn unbuffered auf.

[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "ki-web-mcp"
version = "0.2.0"
requires-python = ">=3.13"
dependencies = [
    "mcp>=1.27,<2",
    "httpx>=0.28,<0.29",
    "pydantic>=2.13,<3",
]

[project.scripts]
ki-web = "ki_web_mcp.server:main"

[tool.setuptools.packages.find]
include = ["ki_web_mcp*"]

5. Domain-Schicht, Daten und Fehler

Nur Datentypen, keine Imports von HTTP, JSON oder mcp-SDK. So bleibt die Domain testbar und ersetzbar.

domain/page.py

from __future__ import annotations
from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True, slots=True)
class Page:
    slug: str
    title: str
    meta_description: str
    content: str
    id: Optional[int] = None
    created_at: Optional[str] = None
    updated_at: Optional[str] = None

domain/errors.py

from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Optional

@dataclass(slots=True)
class BackendError(Exception):
    status: int
    error_code: str
    message: str
    payload: Optional[dict[str, Any]] = None

@dataclass(slots=True)
class TransportError(Exception):
    status: int
    detail: str

6. Application-Schicht, Port und Use Cases

Der Port ist die einzige Schnittstelle, die die Anwendung kennt. Use Cases orchestrieren, keine I/O, kein Mapping.

application/ports.py

from __future__ import annotations
from typing import Optional, Protocol
from ki_web_mcp.domain.page import Page

class PagesPort(Protocol):
    def list(self, search, sort, direction, limit, offset) -> dict: ...
    def get(self, slug: str) -> Page: ...
    def create(self, slug, title, meta_description, content) -> Page: ...
    def update(self, slug, *, title, meta_description, content) -> Page: ...
    def delete(self, slug: str) -> dict: ...
    def ping(self) -> bool: ...
    def close(self) -> None: ...

application/use_cases.py

from dataclasses import asdict
from ki_web_mcp.application.ports import PagesPort
from ki_web_mcp.domain.page import Page

class PageUseCases:
    def __init__(self, pages: PagesPort):
        self._pages = pages

    def list_pages(self, *, search, sort, direction, limit, offset):
        r = self._pages.list(search, sort, direction, limit, offset)
        return {"total": r["total"], "items": r["items"]}

    def get_page(self, slug):
        return asdict(self._pages.get(slug))

    def create_page(self, slug, title, meta_description, content):
        return asdict(self._pages.create(slug, title, meta_description, content))

    def update_page(self, slug, *, title, meta_description, content):
        return asdict(self._pages.update(
            slug, title=title, meta_description=meta_description, content=content))

    def delete_page(self, slug):
        return self._pages.delete(slug)

7. Infrastructure, HTTP-Adapter und Mapper

Hier passiert die einzige I/O. Der Adapter implementiert den Port, mappt JSON auf Domain-Objekte, kümmert sich um Retries, atexit-Cleanup und Request-Id-Propagation.

infrastructure/page_mapper.py

from ki_web_mcp.domain.page import Page

class PageMapper:
    @staticmethod
    def from_api(data: dict) -> Page:
        return Page(
            slug=data["slug"], title=data["title"],
            meta_description=data["meta_description"],
            content=data["content"],
            id=data.get("id"),
            created_at=data.get("created_at"),
            updated_at=data.get("updated_at"),
        )

infrastructure/pages_client.py, Auszug

import atexit, uuid, logging
import httpx
from ki_web_mcp.domain.errors import BackendError, TransportError
from ki_web_mcp.infrastructure.page_mapper import PageMapper

log = logging.getLogger(__name__)

class PagesClient:
    def __init__(self, base_url: str, timeout_seconds: float, retries: int):
        self._client = httpx.Client(
            base_url=base_url,
            timeout=timeout_seconds,
            headers={"Accept": "application/json"},
            follow_redirects=True,
            transport=httpx.HTTPTransport(retries=retries),
        )
        atexit.register(self._client.close)

    def _request(self, method, path, **kw):
        request_id = uuid.uuid4().hex[:12]
        headers = kw.pop("headers", {}) | {"X-Request-Id": request_id}
        try:
            r = self._client.request(method, path, headers=headers, **kw)
        except httpx.HTTPError as exc:
            raise TransportError(599, f"{type(exc).__name__}: {exc}") from exc
        ct = r.headers.get("content-type", "")
        if not ct.startswith("application/json"):
            raise TransportError(r.status_code, f"non-JSON content_type={ct!r}")
        payload = r.json()
        if r.status_code >= 400:
            raise BackendError(
                status=r.status_code,
                error_code=str(payload.get("error", "unknown")),
                message=str(payload.get("message", "")),
                payload=payload,
            )
        return payload

Wichtige Disziplin: bei nicht-JSON-Antworten (Reverse-Proxy-HTML-Errors, Caddy 502-Seiten) wird der Body NICHT in die Tool-Antwort gereicht, sonst sieht das LLM Stack-Traces.

8. Schemas, Literal, Annotated, TypedDict

Diese Datei ist der Vertrag zwischen Tool und LLM. Das mcp-SDK liest die Type-Hints und erzeugt daraus das JSON-Schema, das die KI vor jedem Aufruf sieht.

from __future__ import annotations
from typing import Annotated, Literal, Optional, TypedDict
from pydantic import Field

SortField = Literal["slug", "title", "meta_description", "updated_at", "created_at"]
SortDirection = Literal["asc", "desc"]
Limit = Annotated[int, Field(ge=1, le=200)]
Offset = Annotated[int, Field(ge=0)]

class PageDict(TypedDict, total=False):
    id: Optional[int]
    slug: str
    title: str
    meta_description: str
    content: str
    created_at: Optional[str]
    updated_at: Optional[str]

class PageListResult(TypedDict):
    total: int
    items: list[PageDict]

class DeleteResult(TypedDict):
    deleted: str

Mit dieser Datei kann das LLM bei sort nur noch die fünf erlaubten Werte schicken, und limit bekommt automatisch eine Schranke 1..200.

9. Composition-Root + server.py

Composition ist die einzige Stelle, an der konkrete Klassen verdrahtet werden. Tests können sie mit Fake-Ports überschreiben.

composition.py

from typing import Optional
from ki_web_mcp.application.ports import PagesPort
from ki_web_mcp.application.use_cases import PageUseCases
from ki_web_mcp.config import Config
from ki_web_mcp.infrastructure.pages_client import PagesClient

class Composition:
    def __init__(self, config: Config, pages: Optional[PagesPort] = None):
        self.config = config
        self._pages = pages or PagesClient(
            base_url=config.api_base,
            timeout_seconds=config.timeout_seconds,
            retries=config.retries,
        )
        self.use_cases = PageUseCases(self._pages)

    def close(self):
        self._pages.close()

server.py, Tools registrieren

import logging, sys
from typing import Optional
from ki_web_mcp.composition import Composition
from ki_web_mcp.config import Config
from ki_web_mcp.schemas import (
    DeleteResult, Limit, Offset, PageDict, PageListResult,
    SortDirection, SortField,
)

def _configure_logging(level: str) -> None:
    logging.basicConfig(stream=sys.stderr,
        level=getattr(logging, level.upper(), logging.WARNING),
        format="%(asctime)s %(levelname)s %(name)s %(message)s")

def build_server(composition):
    from mcp.server.fastmcp import FastMCP
    mcp = FastMCP("ki-web")
    uc = composition.use_cases

    @mcp.tool()
    def list_pages(search: str = "",
                   sort: SortField = "updated_at",
                   direction: SortDirection = "desc",
                   limit: Limit = 50,
                   offset: Offset = 0) -> PageListResult:
        '''List pages with optional search, sort and pagination.'''
        return uc.list_pages(search=search or None, sort=sort,
                             direction=direction, limit=limit, offset=offset)

    @mcp.tool()
    def update_page(slug: str,
                    title: Optional[str] = None,
                    meta_description: Optional[str] = None,
                    content: Optional[str] = None) -> PageDict:
        '''Partial update, provide only the fields you want to change.'''
        return uc.update_page(slug, title=title,
                              meta_description=meta_description, content=content)

    # ... get_page, create_page, delete_page analog
    return mcp

def main() -> int:
    if "--health" in sys.argv:
        cfg = Config.from_env()
        _configure_logging(cfg.log_level)
        c = Composition(cfg)
        try:
            ok = c.use_cases.ping()
        finally:
            c.close()
        print("ok" if ok else "fail", file=sys.stderr)
        return 0 if ok else 1
    cfg = Config.from_env()
    _configure_logging(cfg.log_level)
    c = Composition(cfg)
    try:
        build_server(c).run()
    finally:
        c.close()
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Beachte die Goldene stdio-Regel: logging.basicConfig(stream=sys.stderr) muss vor jedem Tool-Aufruf konfiguriert sein, sonst landet ein Library-Log auf stdout und korrumpiert die JSON-RPC-Frames.

10. Wrapper-Script und Installation

/usr/local/bin/ki-web

#!/bin/sh
exec env PYTHONUNBUFFERED=1 /opt/venv/main/bin/python -u -m ki_web_mcp.server "$@"

Beide Schalter zusammen, PYTHONUNBUFFERED=1 und python -u, verhindern, dass Python stdout/stderr puffert.

Installation und Lockfile

chmod 0755 /usr/local/bin/ki-web
/opt/venv/main/bin/pip install -e /var/www/mcp
/opt/venv/main/bin/pip freeze | sort > /var/www/mcp/requirements.lock

11. Tests, drei Schichten

conftest.py, FakePages-Port

import pytest
from ki_web_mcp.application.ports import PagesPort
from ki_web_mcp.application.use_cases import PageUseCases
from ki_web_mcp.domain.errors import BackendError
from ki_web_mcp.domain.page import Page

class FakePages(PagesPort):
    def __init__(self):
        self._store: dict[str, Page] = {}
    def create(self, slug, title, meta, content):
        if slug in self._store:
            raise BackendError(409, "already_exists", slug)
        p = Page(slug=slug, title=title, meta_description=meta, content=content)
        self._store[slug] = p
        return p
    # list/get/update/delete/ping/close analog
    def close(self): pass

@pytest.fixture
def use_cases(): return PageUseCases(FakePages())

test_use_cases.py, Beispiel

import pytest
from ki_web_mcp.domain.errors import BackendError

def test_create_then_get_roundtrip(use_cases):
    use_cases.create_page("foo", "Foo", "Meta", "<h1>x</h1>")
    page = use_cases.get_page("foo")
    assert page["slug"] == "foo"

def test_create_duplicate_raises_already_exists(use_cases):
    use_cases.create_page("foo", "Foo", "Meta", "<p>x</p>")
    with pytest.raises(BackendError) as exc:
        use_cases.create_page("foo", "Foo", "Meta", "<p>x</p>")
    assert exc.value.status == 409

test_schema.py, Subprocess-Roundtrip

Stellt sicher, dass jeder Tool-Name und jedes Schema-Detail (Literal-Enums, Range-Bounds, Optional-Felder bei update_page) erhalten bleibt:

import json, subprocess, sys

def _send(p, m): p.stdin.write(json.dumps(m) + "\n"); p.stdin.flush()
def _rpc(p, m):
    _send(p, m); return json.loads(p.stdout.readline())

def test_tools_list_matches_contract():
    p = subprocess.Popen(
        [sys.executable, "-u", "-m", "ki_web_mcp.server"],
        stdin=subprocess.PIPE, stdout=subprocess.PIPE,
        stderr=subprocess.PIPE, text=True,
    )
    try:
        _rpc(p, {"jsonrpc":"2.0","id":1,"method":"initialize",
                 "params":{"protocolVersion":"2025-03-26",
                           "capabilities":{},
                           "clientInfo":{"name":"t","version":"0"}}})
        _send(p, {"jsonrpc":"2.0","method":"notifications/initialized"})
        resp = _rpc(p, {"jsonrpc":"2.0","id":2,"method":"tools/list"})
    finally:
        try: p.stdin.close()
        except BrokenPipeError: pass
        try: p.wait(timeout=2)
        except subprocess.TimeoutExpired: p.kill(); p.wait(timeout=2)
    tools = {t["name"]: t for t in resp["result"]["tools"]}
    assert set(tools) == {"list_pages","get_page","create_page",
                          "update_page","delete_page"}
    assert "enum" in json.dumps(tools["list_pages"]["inputSchema"]["properties"]["sort"])

Wichtig: Notifikationen (notifications/initialized) liefern KEINE Antwort, also nicht readline() darauf anwenden, sonst blockt der Test.

12. MCP-Clients verbinden

Schreibweise je CLI mit den jeweiligen kleinen Unterschieden:

# Claude Code, separator vor dem Command
claude mcp add ki-web -- /usr/local/bin/ki-web

# Codex, ebenfalls separator
codex mcp add ki-web -- /usr/local/bin/ki-web

# qwen-code, KEIN separator
qwen mcp add ki-web /usr/local/bin/ki-web

Verifizieren mit claude mcp list, codex mcp list bzw. qwen mcp list, der Eintrag sollte überall als connected auftauchen.

13. Debugging

Manueller stdio-Roundtrip

Wenn ein Client meldet, der Server bricht ab, fährt ein Hand-Test schneller zur Diagnose als Logs:

printf '%s\n' \
  '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"t","version":"0"}}}' \
  '{"jsonrpc":"2.0","method":"notifications/initialized"}' \
  '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' \
  | /usr/local/bin/ki-web | jq .

Healthcheck

/usr/local/bin/ki-web --health   # exit 0/1, Trockenlauf gegen /api/pages

Schema-Drift

Wenn das Backend ein Feld umbenennt, fällt der pytest test_schema sofort um, Pflicht-Bestandteil jedes CI-Laufs.

stdout-Pollution finden

Wenn die JSON-RPC-Antwort kaputt ankommt, ist meistens ein Library-Log das Problem. Test: vom CLI-Wrapper aus print() oder Log-Output durch dieses Kommando schicken, alles, was nicht mit { beginnt, ist Pollution:

/usr/local/bin/ki-web < /dev/null 2>/dev/null | head -5
# Erwartet: nichts (Server wartet auf Input). Kommt da Text raus,
# stammt er aus stdout-Pollution einer Dependency.

14. Weiterlesen

/mcp-ki-web, Architektur-Sicht: was passiert wo, wer kennt wen
/mcp-server-erstellen-regeln, die Regeln, die diese Anleitung umsetzt
/var/www/code/python-mcp/rules.yaml, maschinenlesbare Variante (25 Regeln)
modelcontextprotocol.io, Protokoll-Spec
github.com/modelcontextprotocol/python-sdk, Python-SDK