帮你快速理解、总结文档立即下载
文档中心>WorkBuddy Enterprise>CodeBuddy>CodeBuddy Code>SDK>SDK MCP 集成

SDK MCP 集成

最近更新时间：2026-08-26 17:59:31
我的收藏

本页目录：

版本要求：本文档针对 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 服务器

配置 MCP 服务器

mcpServers 支持两种配置方式：
1. 对象格式：直接在代码中定义服务器配置
2. 文件路径字符串：指向一个 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 配置

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 服务器通过 HTTP 流式传输与远程服务通信。
{
  type: 'http',
  url: 'https://mcp.example.com/api/v1',
  headers: {
    'Authorization': 'Bearer your-token',
    'Content-Type': 'application/json'
  }
}

SSE 配置

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 工具

配置完 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 服务器

以下是一个完整的示例，展示如何创建和使用数据库 MCP 服务器。

创建 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()

在 SDK 中使用

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())

实例：API 集成 MCP 服务器

以下示例展示如何创建一个通过 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()

在 SDK 中使用

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 服务器

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 概览
TypeScript SDK 参考
Python SDK 参考
MCP 使用文档
SDK 权限系统
SDK Hooks 系统

更多资源

MCP 官方文档
MCP Python SDK - FastMCP
MCP TypeScript SDK
中国站