MCP 补全 API

补全（Completion）API 是 MCP 提供的一项实用功能，允许服务器为提示模板参数和资源 URI 提供自动补全建议。这可以显著提升用户体验，特别是在需要用户输入复杂参数值时。

什么是补全 API？

补全 API 让 MCP 服务器能够像 IDE 的代码补全一样，为用户提供智能的参数建议。当用户在填写提示模板参数或输入资源 URI 时，服务器可以根据当前已输入的内容返回匹配的建议列表。

典型应用场景

• 提示模板参数补全：用户选择"代码审查"提示模板时，补全可用的文件路径
• 资源 URI 补全：用户输入资源 URI 时，补全可用的资源名称
• 动态选项补全：根据上下文提供相关的选项建议

工作原理

补全 API 采用请求-响应模式，客户端发送当前输入状态，服务器返回匹配的建议列表。

能力声明

支持补全功能的服务器必须在初始化时声明 completions 能力：

from mcp.types import ServerCapabilities

# 声明支持补全
server_capabilities = ServerCapabilities(
    completions={}  # 声明支持补全功能
)

使用 FastMCP 时，能力声明会自动处理。

协议消息

请求格式

客户端通过 completion/complete 方法请求补全建议：

{
  "jsonrpc": "2.0",
  "id": "comp-1",
  "method": "completion/complete",
  "params": {
    "ref": {
      "type": "ref/prompt",
      "name": "code_review"
    },
    "argument": {
      "name": "file_path",
      "value": "src/"
    },
    "context": {
      "arguments": {
        "language": "python"
      }
    }
  }
}

响应格式

服务器返回补全建议列表：

{
  "jsonrpc": "2.0",
  "id": "comp-1",
  "result": {
    "completion": {
      "values": [
        "src/main.py",
        "src/utils.py",
        "src/config.py"
      ],
      "total": 15,
      "hasMore": true
    }
  }
}

字段说明

请求参数：

字段	类型	说明
ref	object	引用对象，指定补全的目标
argument.name	string	参数名称
argument.value	string	当前已输入的值
context.arguments	object	已解析的其他参数值

响应字段：

字段	类型	说明
values	string[]	补全建议列表，最多 100 项
total	number	可选，总匹配数
hasMore	boolean	是否有更多结果

引用类型

补全 API 支持两种引用类型：

1. 提示模板引用 (ref/prompt)

为提示模板的参数提供补全：

{
  "ref": {
    "type": "ref/prompt",
    "name": "code_review"
  },
  "argument": {
    "name": "file_path",
    "value": "src/"
  }
}

2. 资源引用 (ref/resource)

为资源 URI 提供补全：

{
  "ref": {
    "type": "ref/resource",
    "uri": "file:///{path}"
  },
  "argument": {
    "name": "path",
    "value": "proje"
  }
}

Python 实现

基础实现

from mcp.server import Server
from mcp.types import CompletionArgument, CompletionContext

app = Server("my-server")

@app.complete()
async def handle_completion(
    ref: dict,
    argument: CompletionArgument,
    context: CompletionContext | None
) -> dict:
    """处理补全请求"""
    
    # 提示模板参数补全
    if ref.get("type") == "ref/prompt":
        prompt_name = ref.get("name")
        arg_name = argument.name
        current_value = argument.value
        
        if prompt_name == "code_review" and arg_name == "file_path":
            # 返回匹配的文件路径
            files = find_matching_files(current_value)
            return {
                "completion": {
                    "values": files[:100],
                    "total": len(files),
                    "hasMore": len(files) > 100
                }
            }
    
    # 资源 URI 补全
    if ref.get("type") == "ref/resource":
        uri_template = ref.get("uri")
        # 根据 URI 模板返回匹配的资源
        resources = match_resources(uri_template, argument.value)
        return {
            "completion": {
                "values": resources[:100],
                "hasMore": len(resources) > 100
            }
        }
    
    # 无匹配时返回空列表
    return {"completion": {"values": [], "hasMore": False}}

完整示例：文件系统补全

from pathlib import Path
from mcp.server import Server
from mcp.types import CompletionArgument, CompletionContext

app = Server("filesystem-server")

# 允许访问的根目录
ALLOWED_ROOTS = [
    Path.home() / "projects",
    Path.home() / "documents"
]

def find_files(prefix: str, base_dirs: list[Path]) -> list[str]:
    """根据前缀查找匹配的文件"""
    results = []
    
    for base_dir in base_dirs:
        if not base_dir.exists():
            continue
        
        # 递归搜索所有文件
        for file_path in base_dir.rglob("*"):
            relative = str(file_path.relative_to(base_dir))
            if relative.startswith(prefix):
                results.append(relative)
    
    return sorted(results)[:100]

@app.complete()
async def handle_completion(
    ref: dict,
    argument: CompletionArgument,
    context: CompletionContext | None
) -> dict:
    """处理文件路径补全"""
    
    if ref.get("type") == "ref/prompt":
        prompt_name = ref.get("name")
        
        # 为"读取文件"提示提供路径补全
        if prompt_name == "read_file":
            current = argument.value
            files = find_files(current, ALLOWED_ROOTS)
            
            return {
                "completion": {
                    "values": files,
                    "hasMore": False
                }
            }
    
    if ref.get("type") == "ref/resource":
        uri_template = ref.get("uri", "")
        
        # 为 file:// 资源提供补全
        if uri_template.startswith("file://"):
            current = argument.value
            files = find_files(current, ALLOWED_ROOTS)
            
            return {
                "completion": {
                    "values": files,
                    "hasMore": False
                }
            }
    
    return {"completion": {"values": [], "hasMore": False}}

使用上下文增强补全

context.arguments 包含了已解析的其他参数值，可以用来提供更智能的补全建议：

@app.complete()
async def handle_completion(
    ref: dict,
    argument: CompletionArgument,
    context: CompletionContext | None
) -> dict:
    
    if ref.get("type") == "ref/prompt":
        prompt_name = ref.get("name")
        
        if prompt_name == "search_code":
            # context.arguments 包含已填写的其他参数
            language = context.arguments.get("language", "") if context else ""
            query = argument.value
            
            # 根据语言过滤文件
            extensions = {
                "python": ".py",
                "javascript": ".js",
                "java": ".java"
            }
            
            ext = extensions.get(language, "")
            files = [f for f in find_files(query) if f.endswith(ext)]
            
            return {
                "completion": {
                    "values": files[:100],
                    "hasMore": len(files) > 100
                }
            }
    
    return {"completion": {"values": [], "hasMore": False}}

FastMCP 实现

使用 FastMCP 可以更简洁地实现补全功能：

from mcp.server.fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("my-server")

# 存储项目文件列表
PROJECT_FILES = [
    "src/main.py",
    "src/utils.py",
    "src/config.py",
    "tests/test_main.py",
    "docs/README.md"
]

@mcp.prompt()
def review_file(file_path: str, focus: str = "all") -> str:
    """审查代码文件
    
    Args:
        file_path: 要审查的文件路径
        focus: 审查重点（security、performance、all）
    """
    return f"请审查文件 {file_path}，重点关注 {focus}"

@mcp.completion("ref/prompt", "review_file", "file_path")
def complete_file_path(argument: dict, context: dict | None) -> list[str]:
    """为 review_file 提示的 file_path 参数提供补全"""
    current = argument.get("value", "")
    
    # 过滤匹配的文件
    matches = [f for f in PROJECT_FILES if f.startswith(current)]
    return matches[:20]

@mcp.completion("ref/prompt", "review_file", "focus")
def complete_focus(argument: dict, context: dict | None) -> list[str]:
    """为 review_file 提示的 focus 参数提供补全"""
    return ["security", "performance", "readability", "all"]

实现最佳实践

1. 按相关性排序

返回的建议应该按相关性排序，最匹配的排在最前面：

def sort_by_relevance(suggestions: list[str], query: str) -> list[str]:
    """按相关性排序建议"""
    # 完全匹配的排最前
    exact = [s for s in suggestions if s == query]
    # 开头匹配的排其次
    starts_with = [s for s in suggestions if s.startswith(query) and s != query]
    # 包含匹配的排最后
    contains = [s for s in suggestions if query in s and not s.startswith(query)]
    
    return exact + starts_with + contains

2. 实现模糊匹配

对于更好的用户体验，可以实现模糊匹配：

def fuzzy_match(text: str, query: str) -> bool:
    """简单的模糊匹配实现"""
    text_lower = text.lower()
    query_lower = query.lower()
    
    # 检查 query 的每个字符是否按顺序出现在 text 中
    pos = 0
    for char in query_lower:
        pos = text_lower.find(char, pos)
        if pos == -1:
            return False
        pos += 1
    return True

def fuzzy_filter(suggestions: list[str], query: str) -> list[str]:
    """模糊过滤建议列表"""
    return [s for s in suggestions if fuzzy_match(s, query)]

3. 限制结果数量

始终限制返回的结果数量，最多 100 项：

MAX_RESULTS = 100

@app.complete()
async def handle_completion(ref, argument, context):
    suggestions = generate_suggestions(argument.value)
    
    return {
        "completion": {
            "values": suggestions[:MAX_RESULTS],
            "total": len(suggestions),
            "hasMore": len(suggestions) > MAX_RESULTS
        }
    }

4. 实现速率限制

防止补全请求消耗过多资源：

from collections import defaultdict
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = defaultdict(list)
    
    def is_allowed(self, client_id: str) -> bool:
        now = datetime.now()
        # 清理过期请求
        self.requests[client_id] = [
            t for t in self.requests[client_id]
            if now - t < self.window
        ]
        
        if len(self.requests[client_id]) >= self.max_requests:
            return False
        
        self.requests[client_id].append(now)
        return True

limiter = RateLimiter(max_requests=10, window_seconds=60)

@app.complete()
async def handle_completion(ref, argument, context):
    client_id = get_client_id()  # 从会话获取客户端标识
    
    if not limiter.is_allowed(client_id):
        # 速率限制时返回空结果
        return {"completion": {"values": [], "hasMore": False}}
    
    # 正常处理补全请求
    # ...

客户端实现建议

客户端在调用补全 API 时应遵循以下最佳实践：

1. 防抖处理

用户输入时进行防抖，避免频繁请求：

// TypeScript 示例
let debounceTimer: NodeJS.Timeout | null = null;

async function requestCompletion(value: string) {
  // 取消之前的请求
  if (debounceTimer) {
    clearTimeout(debounceTimer);
  }
  
  // 300ms 后发起请求
  debounceTimer = setTimeout(async () => {
    const result = await client.completion.complete({
      ref: { type: "ref/prompt", name: "my_prompt" },
      argument: { name: "file_path", value }
    });
    displaySuggestions(result.completion.values);
  }, 300);
}

2. 缓存结果

对于相同输入，可以缓存补全结果：

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_completion(prefix: str) -> tuple[str, ...]:
    """缓存补全结果"""
    return tuple(find_files(prefix))

@app.complete()
async def handle_completion(ref, argument, context):
    suggestions = get_cached_completion(argument.value)
    return {
        "completion": {
            "values": list(suggestions[:100]),
            "hasMore": len(suggestions) > 100
        }
    }

3. 优雅处理错误

补全失败时不应影响用户体验：

async function fetchCompletions(value: string) {
  try {
    const result = await client.completion.complete({...});
    return result.completion.values;
  } catch (error) {
    // 静默失败，返回空数组
    console.warn("Completion failed:", error);
    return [];
  }
}

安全考虑

输入验证

验证所有补全请求的输入：

@app.complete()
async def handle_completion(ref, argument, context):
    # 验证引用类型
    if ref.get("type") not in ["ref/prompt", "ref/resource"]:
        return {"completion": {"values": [], "hasMore": False}}
    
    # 验证参数值长度
    if len(argument.value) > 1000:
        return {"completion": {"values": [], "hasMore": False}}
    
    # 验证参数名称
    if not argument.name.isidentifier():
        return {"completion": {"values": [], "hasMore": False}}
    
    # 正常处理...

访问控制

确保补全建议不会泄露敏感信息：

@app.complete()
async def handle_completion(ref, argument, context):
    # 检查用户权限
    user = get_current_user()
    if not user.has_permission("read_files"):
        return {"completion": {"values": [], "hasMore": False}}
    
    # 过滤敏感文件
    suggestions = [f for f in find_files(argument.value)
                   if not is_sensitive(f)]
    
    return {"completion": {"values": suggestions[:100]}}

错误处理

服务器应返回标准的 JSON-RPC 错误：

错误码	场景
-32601	不支持补全功能
-32602	无效的提示名称或参数
-32603	内部错误

from mcp.types import McpError

@app.complete()
async def handle_completion(ref, argument, context):
    if ref.get("type") == "ref/prompt":
        prompt_name = ref.get("name")
        
        # 检查提示是否存在
        if not prompt_exists(prompt_name):
            raise McpError(
                code=-32602,
                message=f"Unknown prompt: {prompt_name}"
            )
    
    # ...

下一步

• 提示模板：学习如何定义提示模板
• 资源管理：了解资源 URI 模板
• 调试指南：掌握调试技巧