UseCase MCP API

中文版

create_use_case_mcp_server

from pydantic_resolve.use_case import create_use_case_mcp_server

mcp = create_use_case_mcp_server(
    apps: list[UseCaseAppConfig],
    name: str = "Pydantic-Resolve UseCase API",
) -> "FastMCP"

Creates an MCP server that exposes UseCaseService methods to AI agents via progressive disclosure.

Parameter	Type	Description
apps	list[UseCaseAppConfig]	Application configurations
name	str	MCP server name (default: "Pydantic-Resolve UseCase API")

Returns a configured FastMCP server instance.

mcp = create_use_case_mcp_server(
    apps=[UseCaseAppConfig(name="project", services=[TaskService])]
)
mcp.run(transport="streamable-http", port=8080)

UseCaseAppConfig

from pydantic_resolve.use_case import UseCaseAppConfig

UseCaseAppConfig(
    name: str,
    services: list[type[UseCaseService]],
    description: str | None = None,
    enable_mutation: bool = True,
    context_extractor: Callable | None = None,
)

Parameter	Type	Description
name	str	Application name (required)
services	list[type[UseCaseService]]	List of UseCaseService subclasses (required)
description	str \| None	Application description for AI agents
enable_mutation	bool	Whether mutation methods are visible in MCP (default: True)
context_extractor	Callable \| None	Callback to extract request-scoped context

context_extractor

Optional callback that extracts request-scoped context (e.g. user identity from Authorization header) from the MCP HTTP request. The extracted dict is merged into method kwargs for parameters annotated with FromContext.

Signature: (Context) -> dict | Awaitable[dict], supports both sync and async.

from fastmcp.server.context import Context
from fastmcp.server.dependencies import get_http_headers

def extract_user_context(ctx: Context) -> dict:
    headers = get_http_headers(include={"authorization"})
    auth = headers.get("authorization", "")
    if auth.startswith("Bearer "):
        token = auth[7:]
        return {"user_id": int(token)}
    return {}

apps = [
    UseCaseAppConfig(
        name="project",
        services=[TaskService],
        context_extractor=extract_user_context,
    ),
]

Data flow:

HTTP Request (Authorization: Bearer <token>)
  → FastMCP Context
    → context_extractor(ctx) → {"user_id": 1}
      → call_use_case merges context into kwargs
        → TaskService.get_my_tasks(user_id=1)

Important: get_http_headers() excludes authorization, content-type, and other sensitive headers by default. You must pass include={"authorization"} to receive the Authorization header. When MCP runs via stdio transport (no HTTP request), get_http_headers() returns an empty dict.

UseCaseService

from pydantic_resolve.use_case import UseCaseService
from pydantic_resolve import query, mutation

class MyService(UseCaseService):
    """Service description (used by AI agents)."""

    @query
    async def my_method(cls, param1: int) -> MyDTO:
        """Method description (used by AI agents)."""
        ...

Base class for business service definitions. The BusinessMeta metaclass automatically discovers methods decorated with @query or @mutation and stores them for introspection.

Conventions:

• Methods must be decorated with @query or @mutation (from pydantic_resolve)
• Methods must be async
• Private methods (prefixed with _) and get_tag_name are excluded from discovery
• Docstrings on the class and methods become descriptions visible to AI agents
• Return type annotations are used for SDL type generation

get_tag_name

@classmethod
def get_tag_name(cls) -> str

Returns the class name by default. Override to customize the OpenAPI tag name when using with FastAPI:

class TaskService(UseCaseService):
    @classmethod
    def get_tag_name(cls):
        return "Tasks"

# Usage in FastAPI
@app.get("/tasks", tags=[TaskService.get_tag_name()])

FromContext

from typing import Annotated
from pydantic_resolve.use_case import FromContext

user_id: Annotated[int, FromContext()]

Marker annotation for method parameters that should receive values from context_extractor rather than from the MCP tool's params JSON. This keeps the method signature identical for both FastAPI (parameter passed directly) and MCP (injected from context).

class TaskService(UseCaseService):
    @query
    async def get_my_tasks(
        cls,
        user_id: Annotated[int, FromContext()],
    ) -> list[TaskSummary]:
        ...

• If the context key is present, it is injected into the method call
• If the context key is missing and the parameter has no default, an error is returned
• If the context key is missing and the parameter has a default, the default is used

Progressive Disclosure Tools

The MCP server registers these tools automatically:

Tool	Layer	Description
list_apps	0	Discover available applications
list_services	1	List services in an app
describe_service	2	Get method signatures, parameter schemas, and DTO type definitions
call_use_case	3	Execute a specific method

call_use_case

call_use_case(
    app_name: str,
    service_name: str,
    method_name: str,
    params: str = "{}",
)

Parameter	Type	Description
app_name	str	Application name (from list_apps)
service_name	str	Service name (from list_services)
method_name	str	Method name (from describe_service)
params	str	JSON string with method parameters (default: "{}")

The params string is parsed as JSON and passed as keyword arguments to the method. Parameters annotated with FromContext are injected from the context_extractor result, not from params.