MCP 补全原语：给参数装上自动补全

flowchart LR
    subgraph 服务端原语
        T[Tools 工具<br/>模型可调用]
        R[Resources 资源<br/>模型可读取]
        P[Prompts 提示词<br/>用户可触发]
        C[Completion 补全<br/>辅助填参]
    end
    User((用户)) -->|填提示词参数<br/>拼资源URI| C
    C -.建议值.-> User
    User -->|触发| P
    User -->|订阅| R
    Model((Claude)) -->|调用| T
    Model -->|读取| R

from mcp.server.mcpserver import MCPServer

mcp = MCPServer(name="completion-demo", version="0.1.0")


@mcp.prompt(description="代码评审提示词")
def review_code(language: str, framework: str, code: str) -> str:
    """对给定语言与框架下的代码进行评审。"""
    return (
        f"你是 {framework} 框架的资深 {language} 工程师。"
        f"请评审下面的代码，先指出三个最严重的问题，再给改写建议：\n\n{code}"
    )


@mcp.resource("github://repos/{owner}/{repo}")
def github_repo(owner: str, repo: str) -> str:
    """读取某个 GitHub 仓库的元信息（演示用，返回静态字符串）。"""
    return f"Repository meta: {owner}/{repo}"


if __name__ == "__main__":
    mcp.run(transport="stdio")

uv init mcp-completion-demo && cd mcp-completion-demo
uv add "mcp>=1.2"
uv run completion_server.py

from mcp.types import (
    Completion,
    CompletionArgument,
    CompletionContext,
    PromptReference,
    ResourceTemplateReference,
)


@mcp.completion()
async def handle_completion(
    ref: PromptReference | ResourceTemplateReference,
    argument: CompletionArgument,
    context: CompletionContext | None,
) -> Completion | None:
    """统一处理提示词与资源模板的参数补全。"""

    # —— 1. 提示词参数补全 ——
    if isinstance(ref, PromptReference):
        # 语言参数：静态候选 + 前缀过滤
        if ref.name == "review_code" and argument.name == "language":
            langs = ["python", "typescript", "javascript", "go", "rust", "java"]
            return Completion(
                values=[x for x in langs if x.startswith(argument.value)],
                has_more=False,
            )
        # 框架参数：依赖已填的 language 做上下文补全
        if ref.name == "review_code" and argument.name == "framework":
            chosen = (context.arguments or {}).get("language", "") if context else ""
            by_lang = {
                "python": ["fastapi", "django", "flask", "langchain"],
                "typescript": ["next", "nest", "express", "remix"],
                "rust": ["axum", "actix", "tokio"],
            }
            pool = by_lang.get(chosen, ["fastapi", "next", "django"])
            return Completion(
                values=[x for x in pool if x.startswith(argument.value)],
                has_more=False,
            )

    # —— 2. 资源模板参数补全 ——
    if isinstance(ref, ResourceTemplateReference):
        if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "owner":
            orgs = ["modelcontextprotocol", "anthropics", "vercel", "huggingface"]
            return Completion(values=[o for o in orgs if o.startswith(argument.value)])
        if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
            # 依赖已填的 owner，给出该组织下的仓库候选
            owner = (context.arguments or {}).get("owner", "") if context else ""
            if owner == "modelcontextprotocol":
                repos = ["python-sdk", "typescript-sdk", "specification", "kotlin-sdk"]
                return Completion(values=[r for r in repos if r.startswith(argument.value)])

    # 不认识的 ref，返回 None，客户端会安静地不弹候选
    return None

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import PromptReference, ResourceTemplateReference

server_params = StdioServerParameters(
    command="uv",
    args=["run", "completion_server.py"],
)


async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 1) 提示词参数补全：language 前缀 "py"
            r = await session.complete(
                ref=PromptReference(type="ref/prompt", name="review_code"),
                argument={"name": "language", "value": "py"},
            )
            print("language 候选:", r.completion.values)   # ['python']

            # 2) 上下文感知补全：已知 language=python，补 framework
            r = await session.complete(
                ref=PromptReference(type="ref/prompt", name="review_code"),
                argument={"name": "framework", "value": ""},
                context_arguments={"language": "python"},
            )
            print("framework 候选:", r.completion.values)
            # ['fastapi', 'django', 'flask', 'langchain']

            # 3) 资源模板参数补全：owner 前缀 "model"
            r = await session.complete(
                ref=ResourceTemplateReference(
                    type="ref/resource", uri="github://repos/{owner}/{repo}"
                ),
                argument={"name": "owner", "value": "model"},
            )
            print("owner 候选:", r.completion.values)      # ['modelcontextprotocol']

            # 4) 二级上下文补全：已知 owner，补 repo
            r = await session.complete(
                ref=ResourceTemplateReference(
                    type="ref/resource", uri="github://repos/{owner}/{repo}"
                ),
                argument={"name": "repo", "value": ""},
                context_arguments={"owner": "modelcontextprotocol"},
            )
            print("repo 候选:", r.completion.values)
            # ['python-sdk', 'typescript-sdk', 'specification', 'kotlin-sdk']


if __name__ == "__main__":
    asyncio.run(main())

uv add "mcp[cli]>=1.2"
uv run completion_client.py

language 候选: ['python']
framework 候选: ['fastapi', 'django', 'flask', 'langchain']
owner 候选: ['modelcontextprotocol']
repo 候选: ['python-sdk', 'typescript-sdk', 'specification', 'kotlin-sdk']

npx @modelcontextprotocol/inspector uv run completion_server.py

import pathlib

SUPPORTED_FILE = pathlib.Path(__file__).parent / "SUPPORTED.md"


async def _load_languages() -> list[str]:
    """从文件读取支持的语言列表，每行一个。"""
    if not SUPPORTED_FILE.exists():
        return ["python", "typescript"]
    lines = [ln.strip().lstrip("- ").strip() for ln in SUPPORTED_FILE.read_text().splitlines()]
    return [ln for ln in lines if ln and not ln.startswith("#")]


@mcp.completion()
async def handle_completion(ref, argument, context):
    if isinstance(ref, PromptReference) and ref.name == "review_code":
        if argument.name == "language":
            pool = await _load_languages()          # 动态来源
            matches = [x for x in pool if x.startswith(argument.value)]
            return Completion(
                values=matches[:100],               # 硬上限 100
                total=len(matches),
                has_more=len(matches) > 100,
            )
    return None

# tz_completion_server.py
from mcp.server.mcpserver import MCPServer
from mcp.types import (
    Completion, CompletionArgument, CompletionContext,
    PromptReference, ResourceTemplateReference,
)

mcp = MCPServer(name="tz-completion", version="0.1.0")

TIMEZONES = [
    "Asia/Shanghai", "Asia/Tokyo", "Asia/Singapore",
    "Europe/London", "Europe/Paris", "Europe/Berlin",
    "America/New_York", "America/Los_Angeles", "UTC",
]
COUNTRIES = {"CN": "China", "JP": "Japan", "US": "United States",
             "GB": "United Kingdom", "DE": "Germany", "FR": "France"}


@mcp.prompt(description="时区换算提示词")
def convert_time(timezone: str, source_time: str) -> str:
    return f"请把时间 {source_time} 换算到 {timezone} 时区，给出当天各主要城市的对应时刻。"


@mcp.resource("country://{code}/info")
def country_info(code: str) -> str:
    name = COUNTRIES.get(code.upper(), "Unknown")
    return f"国家代码 {code.upper()} -> {name}"


@mcp.completion()
async def handle_completion(ref, argument: CompletionArgument, context):
    if isinstance(ref, PromptReference) and ref.name == "convert_time" \
            and argument.name == "timezone":
        v = argument.value.lower()
        return Completion(values=[tz for tz in TIMEZONES if tz.lower().startswith(v)])

    if isinstance(ref, ResourceTemplateReference) \
            and ref.uri == "country://{code}/info" and argument.name == "code":
        v = argument.value.upper()
        return Completion(values=[c for c in COUNTRIES if c.startswith(v)])

    return None


if __name__ == "__main__":
    mcp.run(transport="stdio")

sequenceDiagram
    participant U as 用户(客户端UI)
    participant C as MCP Client
    participant S as 补全服务器

    U->>C: 在 timezone 框输入 "asia/"
    C->>S: completion/complete<br/>{ref: convert_time, argument:{timezone:"asia/"}}
    S-->>C: {values:["Asia/Shanghai","Asia/Tokyo","Asia/Singapore"]}
    C-->>U: 下拉显示 3 个候选

    U->>C: 选定 timezone=Asia/Shanghai，继续填 source_time
    Note over C,S: 补全只服务"填参"，不参与模型推理
    U->>C: 触发提示词
    C->>S: prompts/get(name=convert_time, args=...)
    S-->>C: 渲染后的消息