Add action/session level MCP interaction logging

AI_CHANGE_CHECKLIST.md

@@ -70,3 +70,11 @@ Run all of:
 - `make test`
 
 Do not conclude work until all pass, or explicitly report blockers.
+
+## 9) MCP Extension Changes
+
+When adding MCP-facing capabilities:
+- Place MCP implementation under `app/domains/<domain>/` (for this repo: `app/domains/mcp_server/`).
+- Keep tool registration grouped by capability scopes so new routers can be mapped without editing core platform files.
+- Ensure API client wrappers normalize downstream failures to RFC 7807-like error documents.
+- Document standalone runtime scripts and required environment variables.

ARCHITECTURE.md

@@ -301,3 +301,16 @@ Worker scaffold:
 - `property`: property-based invariants
 
 Coverage is enforced at 100% line + branch for all modules under `app/`.
+
+## MCP Extension Domain
+
+- Package: `app/domains/mcp_server/`
+- Purpose: expose grouped MCP tools/resources (`auth`, `users`, `posts`, `vote`) mapped to existing `/api/v1/*` routes.
+- Standalone runtime: `fastapi-template-mcp` (script entrypoint in `pyproject.toml`) starts a dedicated FastAPI process that serves:
+  - `GET /mcp/tools`
+  - `GET /mcp/resources`
+  - `POST /mcp/tools/call`
+  - `GET /mcp/config`
+  - `GET /mcp/health`
+- API calls are delegated through a typed async HTTP client with normalized problem-document errors compatible with RFC 7807-like payloads.
+- MCP tool execution logging includes action-level IDs, session IDs, source metadata, and outbound API call logs for audit/debug timelines.

README.md

@@ -144,3 +144,25 @@ Kubernetes examples:
 - Architecture: `ARCHITECTURE.md`
 - AI checklist: `AI_CHANGE_CHECKLIST.md`
 - AI agent instructions: `AGENTS.md`
+
+## MCP Server Extension (Standalone)
+
+A new extension package is available at `app/domains/mcp_server/` to expose template API capabilities as MCP-style tools/resources without modifying platform core files.
+
+Run standalone MCP server:
+
+```bash
+fastapi-template-mcp
+```
+
+Environment variables (prefix `MCP_SERVER_`):
+- `MCP_SERVER_HOST`
+- `MCP_SERVER_PORT`
+- `MCP_SERVER_BASE_URL`
+- `MCP_SERVER_AUTH_MODE` (`none` or `bearer`)
+- `MCP_SERVER_TIMEOUT_SECONDS`
+- `MCP_SERVER_ALLOWED_TOOL_SCOPES` (CSV list, e.g. `auth,users,posts,vote`)
+
+Default tool groups are mapped from current `/api/v1` endpoints using existing router capability tags (`Authentication`, `Users`, `Post`, `Vote`).
+
+MCP interactions are logged with per-action IDs and session IDs to support traceability for debugging and security reviews.

app/domains/mcp_server/__init__.py

@@ -0,0 +1,5 @@
+"""MCP server domain package."""
+
+from .server import MCPServer, create_mcp_app
+
+__all__ = ["MCPServer", "create_mcp_app"]

app/domains/mcp_server/cli.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import uvicorn
+
+from .config import MCPServerSettings
+from .server import create_mcp_app
+
+
+def main() -> None:
+    settings = MCPServerSettings()
+    app = create_mcp_app(settings)
+    uvicorn.run(app, host=settings.host, port=settings.port)
+
+
+if __name__ == "__main__":
+    main()

app/domains/mcp_server/client.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import logging
+from typing import Any, TypeVar
+
+import httpx
+from pydantic import BaseModel
+
+from .problem import APIClientError, ProblemDocument, normalized_problem
+
+ResponseModelT = TypeVar("ResponseModelT", bound=BaseModel)
+logger = logging.getLogger(__name__)
+
+
+class APIClient:
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        timeout_seconds: float,
+        bearer_token: str | None = None,
+    ) -> None:
+        headers: dict[str, str] = {}
+        if bearer_token:
+            headers["Authorization"] = f"Bearer {bearer_token}"
+        self._client = httpx.AsyncClient(
+            base_url=base_url,
+            timeout=timeout_seconds,
+            headers=headers,
+        )
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    async def request(
+        self,
+        *,
+        method: str,
+        path: str,
+        response_model: type[ResponseModelT],
+        json_body: BaseModel | dict[str, Any] | None = None,
+        params: dict[str, Any] | None = None,
+        action_id: str | None = None,
+        session_id: str | None = None,
+    ) -> ResponseModelT:
+        payload = json_body.model_dump() if isinstance(json_body, BaseModel) else json_body
+        logger.info(
+            "mcp_outbound_request",
+            extra={
+                "action_id": action_id,
+                "session_id": session_id,
+                "method": method,
+                "path": path,
+            },
+        )
+        response = await self._client.request(
+            method=method,
+            url=path,
+            json=payload,
+            params=params,
+            headers={
+                "X-MCP-Action-ID": action_id or "",
+                "X-MCP-Session-ID": session_id or "",
+            },
+        )
+        if response.is_error:
+            self._raise_api_error(
+                path=path,
+                response=response,
+                action_id=action_id,
+                session_id=session_id,
+            )
+        logger.info(
+            "mcp_outbound_response",
+            extra={
+                "action_id": action_id,
+                "session_id": session_id,
+                "status_code": response.status_code,
+                "path": path,
+            },
+        )
+        return response_model.model_validate(response.json())
+
+    @staticmethod
+    def _raise_api_error(
+        *,
+        path: str,
+        response: httpx.Response,
+        action_id: str | None = None,
+        session_id: str | None = None,
+    ) -> None:
+        data = response.json() if response.content else {}
+        if isinstance(data, dict) and all(
+            key in data for key in ("title", "status", "detail", "error_code")
+        ):
+            problem = ProblemDocument.model_validate(data)
+        else:
+            detail = str(data.get("detail", response.text or "Request failed")) if isinstance(data, dict) else (
+                response.text or "Request failed"
+            )
+            problem = normalized_problem(
+                status=response.status_code,
+                detail=detail,
+                instance=path,
+                error_code="request_failed",
+            )
+        logger.warning(
+            "mcp_outbound_error",
+            extra={
+                "action_id": action_id,
+                "session_id": session_id,
+                "status": problem.status,
+                "error_code": problem.error_code,
+                "instance": problem.instance,
+            },
+        )
+        raise APIClientError(problem)

app/domains/mcp_server/config.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from enum import Enum
+
+from pydantic import AnyHttpUrl, Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class MCPAuthMode(str, Enum):
+    NONE = "none"
+    BEARER = "bearer"
+
+
+class MCPServerSettings(BaseSettings):
+    host: str = "127.0.0.1"
+    port: int = 8765
+    base_url: AnyHttpUrl = "http://127.0.0.1:8000"
+    auth_mode: MCPAuthMode = MCPAuthMode.BEARER
+    timeout_seconds: float = 10.0
+    allowed_tool_scopes: list[str] = Field(
+        default_factory=lambda: ["auth", "users", "posts", "vote"]
+    )
+
+    model_config = SettingsConfigDict(
+        env_prefix="MCP_SERVER_",
+        env_file=".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    @field_validator("timeout_seconds")
+    @classmethod
+    def validate_timeout(cls, value: float) -> float:
+        if value <= 0:
+            raise ValueError("MCP_SERVER_TIMEOUT_SECONDS must be > 0")
+        return value
+
+    @field_validator("allowed_tool_scopes", mode="before")
+    @classmethod
+    def parse_scopes(cls, value: object) -> list[str]:
+        if isinstance(value, str):
+            return [scope.strip() for scope in value.split(",") if scope.strip()]
+        if isinstance(value, list):
+            return [str(scope).strip() for scope in value if str(scope).strip()]
+        raise ValueError("MCP_SERVER_ALLOWED_TOOL_SCOPES must be a list or csv string")
+
+
+settings = MCPServerSettings()

app/domains/mcp_server/problem.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from http import HTTPStatus
+
+from pydantic import BaseModel
+
+
+class ProblemDocument(BaseModel):
+    type: str = "about:blank"
+    title: str
+    status: int
+    detail: str
+    instance: str = ""
+    error_code: str = "request_failed"
+
+
+class APIClientError(Exception):
+    def __init__(self, problem: ProblemDocument):
+        self.problem = problem
+        super().__init__(problem.detail)
+
+
+def normalized_problem(*, status: int, detail: str, instance: str, error_code: str) -> ProblemDocument:
+    try:
+        title = HTTPStatus(status).phrase
+    except ValueError:
+        title = "Error"
+    return ProblemDocument(
+        title=title,
+        status=status,
+        detail=detail,
+        instance=instance,
+        error_code=error_code,
+    )