版本要求：本文档针对 CodeBuddy Agent SDK v0.1.24 及以上版本。
功能状态：SDK MCP 支持是 CodeBuddy Agent SDK 的一项 Preview 功能。
本文档介绍如何在 CodeBuddy Agent SDK 中集成和使用 MCP（Model Context Protocol）服务器，为您的应用程序扩展自定义工具和功能。
MCP 服务器允许您将自定义工具、资源和提示集成到 CodeBuddy Agent SDK 中。通过 SDK，您可以以编程方式启用这些工具，从而为您的 AI 工作流添加特定领域的功能。
MCP 服务器：提供工具、资源和提示的独立进程或服务
工具：MCP 服务器暴露的可执行函数
资源：MCP 服务器提供的可读取的数据或文件
提示：MCP 服务器提供的模板化提示词
CodeBuddy Agent SDK 支持三种 MCP 服务器通信方式：
传输类型 | 使用场景 | 说明 |
STDIO | 本地工具 | 通过标准输入输出与本地进程通信 |
HTTP/SSE | 远程服务 | 通过 HTTP 流式传输或 Server-Sent Events 与远程服务通信 |
SDK MCP | IDE/SDK 集成 | 由外部 SDK（如 IDE 扩展）注册的 MCP 服务器 |
mcpServers 支持两种配置方式：
对象格式：直接在代码中定义服务器配置
文件路径字符串：指向一个 MCP 配置文件（JSON 格式）
使用 mcpServers 选项配置 MCP 服务器。配置支持 STDIO、HTTP 和 SSE 传输类型。
TypeScript：
import { query } from '@tencent-ai/agent-sdk';
const result = query({
prompt: 'Analyze my project structure',
options: {
mcpServers: {
'my-tools': {
type: 'stdio',
command: 'node',
args: ['./mcp-server.js'],
env: {
NODE_ENV: 'production'
}
},
'api-server': {
type: 'http',
url: 'https://api.example.com/mcp',
headers: {
'Authorization': 'Bearer your-token'
}
},
'sse-server': {
type: 'sse',
url: 'https://events.example.com/mcp/sse',
headers: {
'X-API-Key': 'your-api-key'
}
}
}
}
});
Python：
from codebuddy_agent_sdk import query
result = query(
prompt='Analyze my project structure',
options={
'mcp_servers': {
'my-tools': {
'type': 'stdio',
'command': 'python',
'args': ['-m', 'my_mcp_server'],
'env': {
'PYTHONPATH': '/path/to/tools'
}
},
'api-server': {
'type': 'http',
'url': 'https://api.example.com/mcp',
'headers': {
'Authorization': 'Bearer your-token'
}
},
'sse-server': {
'type': 'sse',
'url': 'https://events.example.com/mcp/sse',
'headers': {
'X-API-Key': 'your-api-key'
}
}
}
}
)
也可以传入一个 MCP 配置文件路径。配置文件为 JSON 格式，结构与 CLI --mcp-config 参数一致。
TypeScript：
import { query } from '@tencent-ai/agent-sdk';
const result = query({
prompt: 'Analyze my project structure',
options: {
mcpServers: './mcp-config.json'
}
});
Python：
from codebuddy_agent_sdk import query
result = query(
prompt='Analyze my project structure',
options={
'mcp_servers': './mcp-config.json'
}
)
配置文件格式示例（mcp-config.json）：
{ "mcpServers": { "my-tools": { "type": "stdio", "command": "node", "args": ["./mcp-server.js"] } }}
注意：
使用文件路径格式时，不支持 SDK MCP 类型（type: 'sdk'）的服务器配置。SDK MCP 服务器仅在对象格式中可用。
STDIO 服务器通过标准输入输出与本地进程通信，适用于本地工具。
{ type: 'stdio', command: 'python', // 可执行文件或命令 args: ['-m', 'my_mcp_server'], // 命令行参数 env: { // 环境变量 PYTHONPATH: '/path/to/tools', DEBUG: 'true' }}
常见用例：
// Python MCP 服务器
{
type: 'stdio',
command: 'python',
args: ['-m', 'fastmcp']
}
// Node.js MCP 服务器
{
type: 'stdio',
command: 'node',
args: ['./server.js']
}
// 本地二进制文件
{
type: 'stdio',
command: '/usr/local/bin/my-tool',
args: ['--config', '/etc/config.yaml']
}
HTTP 服务器通过 HTTP 流式传输与远程服务通信。
{ type: 'http', url: 'https://mcp.example.com/api/v1', headers: { 'Authorization': 'Bearer your-token', 'Content-Type': 'application/json' }}
SSE 服务器通过 Server-Sent Events 与远程服务通信。
{ type: 'sse', url: 'https://events.example.com/mcp/sse', headers: { 'Authorization': 'Bearer your-token', 'X-API-Version': 'v1' }}
MCP 工具支持精细化的权限控制。通过 canUseTool 回调，您可以决定哪些工具可以被使用。
type PermissionMode =
| 'default' // 询问用户
| 'acceptEdits' // 接受自动编辑
| 'bypassPermissions' // 跳过权限检查（不推荐用于生产环境）
| 'plan'; // 仅规划模式
const result = query({
prompt: 'Process data',
options: {
permissionMode: 'acceptEdits',
canUseTool?: (toolCall) => {
// 自定义权限检查逻辑
if (toolCall.name === 'dangerous_operation') {
return false;
}
return true;
}
}
});
options: { canUseTool: (toolCall) => { // 阻止特定服务器的工具 if (toolCall.name.startsWith('mcp__restricted')) { return false; } // 允许特定工具 if (toolCall.name === 'mcp__github__list_issues') { return true; } // 默认询问 return null; }}
配置完 MCP 服务器后，Agent 会自动发现这些服务器提供的工具，并在需要时调用它们。
const result = query({
prompt: `
Using the available MCP tools, complete these tasks:
1. Query the database for recent transactions
2. Analyze the results
3. Generate a summary report
`,
options: {
mcpServers: {
'database': {
type: 'stdio',
command: 'python',
args: ['-m', 'db_mcp_server']
}
}
}
});
for await (const message of result) {
if (message.type === 'tool_use') {
console.log(`Tool: ${message.name}`);
console.log(`Arguments: ${JSON.stringify(message.input)}`);
}
}
for await (const message of result) { if (message.type === 'content') { console.log('Agent Response:', message.text); } else if (message.type === 'tool_use') { console.log(`Executing: ${message.name}`); } else if (message.type === 'tool_result') { console.log('Tool Result:', message.content); }}
以下是一个完整的示例，展示如何创建和使用数据库 MCP 服务器。
db_mcp_server.py：
from fastmcp import FastMCP
mcp = FastMCP('database')
@mcp.tool()
def query_database(sql: str) -> str:
"""Execute a SQL query and return results"""
# 实现数据库查询逻辑
return f"Query results for: {sql}"
@mcp.tool()
def get_schema(table_name: str) -> str:
"""Get the schema of a specific table"""
# 返回表结构
return f"Schema for: {table_name}"
if __name__ == '__main__':
mcp.run()
TypeScript：
import { query } from '@tencent-ai/agent-sdk';
async function analyzeData() {
const result = query({
prompt: `
I need to analyze our user activity data.
1. First, check the schema of the users table
2. Query for users active in the last 7 days
3. Provide insights based on the results
`,
options: {
mcpServers: {
'database': {
type: 'stdio',
command: 'python',
args: ['-m', 'db_mcp_server'],
env: {
DATABASE_URL: process.env.DATABASE_URL
}
}
},
permissionMode: 'acceptEdits'
}
});
for await (const message of result) {
if (message.type === 'content') {
console.log('Analysis:', message.text);
}
}
}
analyzeData().catch(console.error);
Python
from codebuddy_agent_sdk import query
import os
async def analyze_data():
result = query(
prompt="""
I need to analyze our user activity data.
1. First, check the schema of the users table
2. Query for users active in the last 7 days
3. Provide insights based on the results
""",
options={
'mcp_servers': {
'database': {
'type': 'stdio',
'command': 'python',
'args': ['-m', 'db_mcp_server'],
'env': {
'DATABASE_URL': os.environ.get('DATABASE_URL')
}
}
},
'permission_mode': 'acceptEdits'
}
)
async for message in result:
if message.get('type') == 'content':
print('Analysis:', message.get('text'))
# 运行示例
import asyncio
asyncio.run(analyze_data())
以下示例展示如何创建一个通过 HTTP 连接到多个 API 的 MCP 服务器。
api_mcp_server.py：
from fastmcp import FastMCP
import httpx
import os
mcp = FastMCP('api-gateway')
async def call_api(method: str, path: str, **kwargs):
"""Helper to call APIs with authentication"""
headers = kwargs.pop('headers', {})
headers['Authorization'] = f"Bearer {os.environ.get('API_TOKEN')}"
async with httpx.AsyncClient() as client:
response = await client.request(
method,
f"https://api.example.com{path}",
headers=headers,
**kwargs
)
return response.json()
@mcp.tool()
async def get_github_repos(username: str) -> dict:
"""Get GitHub repositories for a user"""
return await call_api('GET', f'/github/users/{username}/repos')
@mcp.tool()
async def search_issues(query: str, labels: list[str] = None) -> dict:
"""Search for issues in GitHub"""
params = {'q': query}
if labels:
params['labels'] = ','.join(labels)
return await call_api('GET', '/github/search/issues', params=params)
@mcp.tool()
async def create_slack_message(channel: str, text: str) -> dict:
"""Send a message to Slack"""
return await call_api(
'POST',
'/slack/chat.postMessage',
json={'channel': channel, 'text': text}
)
if __name__ == '__main__':
mcp.run()
import { query } from '@tencent-ai/agent-sdk';
async function searchAndNotify() {
const result = query({
prompt: `
1. Search for open issues labeled "bug" and "critical"
2. Get my GitHub repositories
3. Send a summary to the #dev-team Slack channel
`,
options: {
mcpServers: {
'api-gateway': {
type: 'http',
url: 'https://mcp-gateway.example.com/api',
headers: {
'X-API-Key': process.env.API_GATEWAY_KEY,
'X-Request-ID': `req-${Date.now()}`
}
}
},
permissionMode: 'acceptEdits'
}
});
for await (const message of result) {
console.log(message);
}
}
SSE 服务器适用于需要实时数据流或事件推送的场景。
import { query } from '@tencent-ai/agent-sdk';
async function monitorSystem() {
const result = query({
prompt: 'Monitor system metrics and alert if any threshold is exceeded',
options: {
mcpServers: {
'monitoring': {
type: 'sse',
url: 'https://monitor.example.com/mcp/events',
headers: {
'Authorization': `Bearer ${process.env.MONITOR_TOKEN}`,
'X-Client-ID': 'codebuddy-sdk'
}
}
}
}
});
for await (const message of result) {
if (message.type === 'content') {
console.log('Alert:', message.text);
}
}
}
在处理 MCP 服务器连接时，监控初始化状态并识别失败的连接：
TypeScript：
import { query } from '@tencent-ai/agent-sdk';
const result = query({
prompt: 'Use my MCP tools',
options: {
mcpServers: {
'my-tool': {
type: 'stdio',
command: 'python',
args: ['-m', 'my_mcp_server']
}
}
}
});
for await (const message of result) {
// 在初始化时检查 MCP 服务器状态
if (message.type === 'system' && message.subtype === 'init') {
const failedServers = message.mcp_servers.filter(
s => s.status !== 'connected'
);
if (failedServers.length > 0) {
console.warn('Failed to connect:', failedServers);
}
}
// 处理执行错误
if (message.type === 'result' && message.subtype === 'error_during_execution') {
console.error('Tool execution failed:', message);
}
}
Python：
from codebuddy_agent_sdk import query
result = query(
prompt='Use my MCP tools',
options={
'mcp_servers': {
'my-tool': {
'type': 'stdio',
'command': 'python',
'args': ['-m', 'my_mcp_server']
}
}
}
)
async for message in result:
# 在初始化时检查 MCP 服务器状态
if message.get('type') == 'system' and message.get('subtype') == 'init':
failed_servers = [
s for s in message.get('mcp_servers', [])
if s.get('status') != 'connected'
]
if failed_servers:
print(f'Failed to connect: {failed_servers}')
# 处理执行错误
if message.get('type') == 'result' and message.get('subtype') == 'error_during_execution':
print('Tool execution failed:', message)
SDK 概览
MCP 使用文档
SDK 权限系统
MCP 官方文档
