SDK 自定义工具

import { query, createSdkMcpServer, tool } from '@tencent-ai/agent-sdk';
import { z } from 'zod';

// 创建 MCP 服务器并定义工具
const calculatorServer = createSdkMcpServer('calculator', {
  tools: [
    tool({
      name: 'add',
      description: 'Add two numbers',
      schema: z.object({
        a: z.number().describe('First number'),
        b: z.number().describe('Second number'),
      }),
      handler: async ({ a, b }) => {
        return { result: a + b };
      },
    }),
    tool({
      name: 'multiply',
      description: 'Multiply two numbers',
      schema: z.object({
        a: z.number().describe('First number'),
        b: z.number().describe('Second number'),
      }),
      handler: async ({ a, b }) => {
        return { result: a * b };
      },
    }),
  ],
});

// 在 SDK 中使用自定义工具
const result = query({
  prompt: 'Calculate 15 + 27 and then multiply the result by 3',
  options: {
    mcpServers: {
      'calculator': calculatorServer,
    },
  },
});

for await (const message of result) {
  console.log(message);
}

from codebuddy_agent_sdk import query, create_sdk_mcp_server, tool
from typing import Any

# 定义工具
@tool(
    "add",
    "Add two numbers",
    {"a": float, "b": float}
)
async def add(args: dict[str, Any]) -> dict[str, Any]:
    return {'result': args['a'] + args['b']}

@tool(
    "multiply",
    "Multiply two numbers",
    {"a": float, "b": float}
)
async def multiply(args: dict[str, Any]) -> dict[str, Any]:
    return {'result': args['a'] * args['b']}

# 创建 MCP 服务器并注册工具
calculator_server = create_sdk_mcp_server(
    name='calculator',
    tools=[add, multiply]
)

# 在 SDK 中使用自定义工具
async def calculate():
    result = query(
        prompt='Calculate 15 + 27 and then multiply the result by 3',
        options={
            'mcp_servers': {
                'calculator': calculator_server,
            },
        },
    )

    async for message in result:
        print(message)

# 运行
import asyncio
asyncio.run(calculate())

import { createSdkMcpServer, tool } from '@tencent-ai/agent-sdk';
import { z } from 'zod';

const myServer = createSdkMcpServer('my-tools', {
  tools: [
    tool({
      name: 'my_tool',
      description: 'Description of what the tool does',
      schema: z.object({
        parameter1: z.string().describe('Description of parameter1'),
        parameter2: z.number().optional().describe('Optional parameter'),
      }),
      handler: async (input) => {
        // 实现工具逻辑
        return {
          result: 'Tool output',
          details: input,
        };
      },
    }),
  ],
});

import { createSdkMcpServer, tool } from '@tencent-ai/agent-sdk';
import { z } from 'zod';
import * as fs from 'fs/promises';
import * as path from 'path';

const fileAnalysisServer = createSdkMcpServer('file-analysis', {
  tools: [
    tool({
      name: 'count_lines',
      description: 'Count lines in a file',
      schema: z.object({
        filePath: z.string().describe('Path to the file'),
      }),
      handler: async ({ filePath }) => {
        try {
          const content = await fs.readFile(filePath, 'utf-8');
          const lineCount = content.split('\\n').length;
          return {
            success: true,
            filePath,
            lineCount,
          };
        } catch (error) {
          return {
            success: false,
            error: error instanceof Error ? error.message : 'Unknown error',
          };
        }
      },
    }),
    tool({
      name: 'list_files',
      description: 'List all files in a directory',
      schema: z.object({
        dirPath: z.string().describe('Path to the directory'),
        pattern: z.string().optional().describe('Optional glob pattern'),
      }),
      handler: async ({ dirPath, pattern }) => {
        try {
          const files = await fs.readdir(dirPath);
          
          let filtered = files;
          if (pattern) {
            const minimatch = require('minimatch').minimatch;
            filtered = files.filter(f => minimatch(f, pattern));
          }
          
          return {
            success: true,
            dirPath,
            files: filtered,
            count: filtered.length,
          };
        } catch (error) {
          return {
            success: false,
            error: error instanceof Error ? error.message : 'Unknown error',
          };
        }
      },
    }),
    tool({
      name: 'get_file_info',
      description: 'Get information about a file',
      schema: z.object({
        filePath: z.string().describe('Path to the file'),
      }),
      handler: async ({ filePath }) => {
        try {
          const stats = await fs.stat(filePath);
          return {
            success: true,
            filePath,
            size: stats.size,
            created: stats.birthtime,
            modified: stats.mtime,
            isDirectory: stats.isDirectory(),
            isFile: stats.isFile(),
          };
        } catch (error) {
          return {
            success: false,
            error: error instanceof Error ? error.message : 'Unknown error',
          };
        }
      },
    }),
  ],
});

export default fileAnalysisServer;

from codebuddy_agent_sdk import create_sdk_mcp_server, tool
from typing import Any
import os

@tool(
    "count_lines",
    "Count lines in a file",
    {"file_path": str}
)
async def count_lines(args: dict[str, Any]) -> dict[str, Any]:
    try:
        with open(args['file_path'], 'r') as f:
            line_count = len(f.readlines())
        return {
            'success': True,
            'file_path': args['file_path'],
            'line_count': line_count,
        }
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
        }

@tool(
    "list_files",
    "List all files in a directory",
    {"dir_path": str, "pattern": str}
)
async def list_files(args: dict[str, Any]) -> dict[str, Any]:
    try:
        files = os.listdir(args['dir_path'])
        
        pattern = args.get('pattern')
        if pattern:
            import fnmatch
            files = [f for f in files if fnmatch.fnmatch(f, pattern)]
        
        return {
            'success': True,
            'dir_path': args['dir_path'],
            'files': files,
            'count': len(files),
        }
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
        }

@tool(
    "get_file_info",
    "Get information about a file",
    {"file_path": str}
)
async def get_file_info(args: dict[str, Any]) -> dict[str, Any]:
    try:
        file_path = args['file_path']
        stat = os.stat(file_path)
        return {
            'success': True,
            'file_path': file_path,
            'size': stat.st_size,
            'created': stat.st_ctime,
            'modified': stat.st_mtime,
            'is_file': os.path.isfile(file_path),
            'is_dir': os.path.isdir(file_path),
        }
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
        }

# 创建 MCP 服务器并注册工具
file_analysis_server = create_sdk_mcp_server(
    name='file-analysis',
    tools=[count_lines, list_files, get_file_info]
)

import { createSdkMcpServer, tool } from '@tencent-ai/agent-sdk';
import { z } from 'zod';

const multiToolServer = createSdkMcpServer('multi-tools', {
  tools: [
    tool({
      name: 'tool_one',
      description: 'First tool',
      schema: z.object({ input: z.string() }),
      handler: async ({ input }) => ({ result: `Tool 1: ${input}` }),
    }),
    tool({
      name: 'tool_two',
      description: 'Second tool',
      schema: z.object({ data: z.number() }),
      handler: async ({ data }) => ({ result: `Tool 2: ${data * 2}` }),
    }),
    tool({
      name: 'tool_three',
      description: 'Third tool',
      schema: z.object({
        name: z.string(),
        age: z.number().optional(),
      }),
      handler: async ({ name, age }) => ({
        result: `Tool 3: ${name}, age ${age ?? 'unknown'}`,
      }),
    }),
  ],
});

// 在 SDK 中使用
const result = query({
  prompt: 'Use all the available tools',
  options: {
    mcpServers: {
      'multi-tools': multiToolServer,
    },
  },
});

from codebuddy_agent_sdk import create_sdk_mcp_server, tool
from typing import Any

@tool("tool_one", "First tool", {"input": str})
async def tool_one(args: dict[str, Any]) -> dict[str, Any]:
    return {'result': f"Tool 1: {args['input']}"}

@tool("tool_two", "Second tool", {"data": int})
async def tool_two(args: dict[str, Any]) -> dict[str, Any]:
    return {'result': f"Tool 2: {args['data'] * 2}"}

@tool("tool_three", "Third tool", {"name": str, "age": int})
async def tool_three(args: dict[str, Any]) -> dict[str, Any]:
    age_str = args.get('age', 'unknown')
    return {'result': f"Tool 3: {args['name']}, age {age_str}"}

# 创建 MCP 服务器并注册工具
server = create_sdk_mcp_server(
    name='multi-tools',
    tools=[tool_one, tool_two, tool_three]
)

import { createSdkMcpServer, tool } from '@tencent-ai/agent-sdk';
import { z } from 'zod';

const dataProcessingServer = createSdkMcpServer('data-processing', {
  tools: [
    tool({
      name: 'process_user_data',
      description: 'Process and validate user data',
      schema: z.object({
        userId: z.number().int().positive().describe('User ID'),
        email: z.string().email().describe('User email'),
        tags: z.array(z.string()).describe('User tags'),
        preferences: z.object({
          notifications: z.boolean().default(true),
          theme: z.enum(['light', 'dark', 'auto']).default('auto'),
        }).optional(),
      }),
      handler: async (input) => {
        // input 类型完全由 Zod schema 推断
        // TypeScript 知道所有字段的类型
        const result = {
          userId: input.userId,
          email: input.email,
          tagCount: input.tags.length,
          hasPreferences: !!input.preferences,
        };
        return result;
      },
    }),
  ],
});

from codebuddy_agent_sdk import create_sdk_mcp_server, tool
from typing import Any

# 使用 JSON Schema 进行高级验证
@tool(
    "process_user_data",
    "Process and validate user data",
    {
        "type": "object",
        "properties": {
            "user_id": {"type": "integer", "minimum": 1},
            "email": {"type": "string", "format": "email"},
            "tags": {"type": "array", "items": {"type": "string"}},
            "notifications": {"type": "boolean", "default": True},
            "theme": {"type": "string", "enum": ["light", "dark", "auto"], "default": "auto"}
        },
        "required": ["user_id", "email", "tags"]
    }
)
async def process_user_data(args: dict[str, Any]) -> dict[str, Any]:
    return {
        'user_id': args['user_id'],
        'email': args['email'],
        'tag_count': len(args['tags']),
        'theme': args.get('theme', 'auto'),
        'notifications': args.get('notifications', True),
    }

# 创建 MCP 服务器并注册工具
server = create_sdk_mcp_server(
    name='data-processing',
    tools=[process_user_data]
)