SDK MCP Integration Guide

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

TypeScript

使用 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

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'
                }
            }
        }
    }
)

服务器配置详解

STDIO 配置

STDIO 服务器通过标准输入输出与本地进程通信，适用于本地工具。

typescript

{
  type: 'stdio',
  command: 'python',                    // 可执行文件或命令
  args: ['-m', 'my_mcp_server'],       // 命令行参数
  env: {                               // 环境变量
    PYTHONPATH: '/path/to/tools',
    DEBUG: 'true'
  }
}

常见用例：

typescript

// 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 流式传输与远程服务通信。

typescript

{
  type: 'http',
  url: 'https://mcp.example.com/api/v1',
  headers: {
    'Authorization': 'Bearer your-token',
    'Content-Type': 'application/json'
  }
}

SSE 配置

SSE 服务器通过 Server-Sent Events 与远程服务通信。

typescript

{
  type: 'sse',
  url: 'https://events.example.com/mcp/sse',
  headers: {
    'Authorization': 'Bearer your-token',
    'X-API-Version': 'v1'
  }
}

权限管理

MCP 工具支持精细化的权限控制。通过 canUseTool 回调，你可以决定哪些工具可以被使用。

工具权限模式

typescript

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;
    }
  }
});

特定工具的权限控制

typescript

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 会自动发现这些服务器提供的工具，并在需要时调用它们。

自动工具发现

typescript

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)}`);
  }
}

处理工具结果

typescript

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：

python

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：

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：

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：

python

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 中使用

typescript

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 服务器适用于需要实时数据流或事件推送的场景。

typescript

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：

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：

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 Integration Guide

概述

核心概念

支持的传输类型

配置 MCP 服务器

TypeScript

Python

服务器配置详解

STDIO 配置

HTTP 配置

SSE 配置

权限管理

工具权限模式

特定工具的权限控制

使用 MCP 工具

自动工具发现

处理工具结果

实例：数据库查询 MCP 服务器

创建 MCP 服务器

在 SDK 中使用

实例：API 集成 MCP 服务器

服务器实现

在 SDK 中使用

实例：远程 SSE 服务器

错误处理

检查服务器初始化状态

相关文档

更多资源

SDK MCP Integration Guide ​

概述 ​

核心概念 ​

支持的传输类型 ​

配置 MCP 服务器 ​

TypeScript ​

Python ​

服务器配置详解 ​

STDIO 配置 ​

HTTP 配置 ​

SSE 配置 ​

权限管理 ​

工具权限模式 ​

特定工具的权限控制 ​

使用 MCP 工具 ​

自动工具发现 ​

处理工具结果 ​

实例：数据库查询 MCP 服务器 ​

创建 MCP 服务器 ​

在 SDK 中使用 ​

实例：API 集成 MCP 服务器 ​

服务器实现 ​

在 SDK 中使用 ​

实例：远程 SSE 服务器 ​

错误处理 ​

检查服务器初始化状态 ​

相关文档 ​

更多资源 ​

SDK MCP Integration Guide

概述

核心概念

支持的传输类型

配置 MCP 服务器

TypeScript

Python

服务器配置详解

STDIO 配置

HTTP 配置

SSE 配置

权限管理

工具权限模式

特定工具的权限控制

使用 MCP 工具

自动工具发现

处理工具结果

实例：数据库查询 MCP 服务器

创建 MCP 服务器

在 SDK 中使用

实例：API 集成 MCP 服务器

服务器实现

在 SDK 中使用

实例：远程 SSE 服务器

错误处理

检查服务器初始化状态

相关文档

更多资源