MCP代理SDK

用于以编程方式执行AI代理任务的Python SDK 自动验证 和 人在循环 控制。

特性

代理执行 --启动 codebuddy CLI代理作为受管子流程
自动验证 --使用自定义函数验证任务结果；验证失败会触发自动重试
循环中的人类 --代理人可以致电 Block 发出需要人为干预的任务信号
生命周期挂钩 --拦截和控制代理行为 PreToolUse, PostToolUse, Stop 以及其他钩事件
工具权限控制 --通过以下方式在运行时批准或拒绝工具调用 can_use_tool 带有可选输入修改的回调
结构化流媒体 --接收键入的事件(AssistantMessage, SystemMessage, AgentResult)via AsyncIterator[StreamEvent]
错误诊断 --进程崩溃会引发stderr捕获和退出代码的特定异常
并发代理 --同时运行多个代理，每个代理由一个唯一的运行ID跟踪
超时支持 --设置每次运行超时，自动阻止超时代理

安装

pip install mcp-agent-sdk

或者从源代码安装：

git clone 
cd MCPAgentSDK
pip install -e .

先决条件

Python≥3.10
codebuddy CLI已安装并可用于 PATH

快速开始

import asyncio
from mcp_agent_sdk import (
    MCPAgentSDK, AgentRunConfig, AgentResult,
    AssistantMessage, SystemMessage, TextBlock,
    AgentStartupError, AgentProcessError,
)

async def main():
    sdk = MCPAgentSDK()
    await sdk.init()

    config = AgentRunConfig(
        prompt="Create a file called hello.txt containing 'Hello, World!'",
    )

    try:
        async for event in sdk.run_agent(config):
            if isinstance(event, AgentResult):
                print(f"Done: {event.status} — {event.message}")
            elif isinstance(event, AssistantMessage):
                for block in event.content:
                    if isinstance(block, TextBlock):
                        print(f"[assistant] {block.text}")
            elif isinstance(event, SystemMessage):
                print(f"[system:{event.subtype}] {event.data}")
    except AgentStartupError as e:
        print(f"Startup failed: {e} (stderr={e.stderr}, exit_code={e.exit_code})")
    except AgentProcessError as e:
        print(f"Process crashed: {e} (stderr={e.stderr}, exit_code={e.exit_code})")

    await sdk.shutdown()

asyncio.run(main())

API 参考

`MCPAgentSDK`

运行代理的主要入口点。

sdk = MCPAgentSDK()
await sdk.init(host="127.0.0.1", port=0)  # port=0 auto-selects

属性	类型	描述
`port`	`int`	MCP服务器绑定到的实际端口
`mcp_server_url`	`str`	MCP端点的完整URL

方法	说明
`init(host, port)`	启动内部MCP服务器
`shutdown()`	停止服务器并清理资源
`run_agent(config)`	运行代理；回报 `AsyncIterator[StreamEvent]`

`AgentRunConfig`

单个代理运行的配置数据类。

@dataclass
class AgentRunConfig:
    prompt: str                                                   # Task description
    validate_fn: Callable[[str], tuple[bool, str]] | None = None  # Custom validator
    on_complete: Callable[[str], None] | None = None              # Success callback
    on_block: Callable[[str], None] | None = None                 # Block callback
    max_retries: int = 3                                          # Validation retry limit
    model: str | None = None                                      # LLM model override
    permission_mode: str = "bypassPermissions"                    # CLI permission mode
    cwd: str | None = None                                        # Working directory
    allowed_tools: list[str] | None = None                        # Restrict agent tools
    mcp_servers: dict[str, Any] = field(default_factory=dict)     # Extra MCP servers to inject
    cli_path: str = "codebuddy"                                   # CLI executable name or path
    extra_args: dict[str, str | None] = field(default_factory=dict)  # Extra CLI flags
    timeout: float | None = None                                  # Timeout in seconds
    hooks: dict[HookEvent, list[HookMatcher]] | None = None       # Lifecycle hooks
    can_use_tool: CanUseTool | None = None                         # Tool permission callback

流事件类型

run_agent() 产量 StreamEvent 子类。使用 isinstance 处理每种类型：

`AssistantMessage`

LLM响应包含键入的内容块。

@dataclass
class AssistantMessage(StreamEvent):
    content: list[ContentBlock]   # TextBlock, ThinkingBlock, ToolUseBlock, ToolResultBlock
    session_id: str = ""

`SystemMessage`

系统事件（初始化、成本等）。

@dataclass
class SystemMessage(StreamEvent):
    subtype: str = ""             # "init", "cost", "error", etc.
    data: dict[str, Any] = {}

`ResultMessage`

会话结束元数据（内部消耗，未产生）。

@dataclass
class ResultMessage(StreamEvent):
    session_id: str = ""
    cost_usd: float = 0.0
    duration_ms: int = 0
    is_error: bool = False
    num_turns: int = 0

`AgentResult`

代理运行的最终结果。

@dataclass
class AgentResult(StreamEvent):
    status: str = ""              # "completed" | "blocked"
    message: str = ""             # Result description
    session_id: str = ""
    agent_run_id: str = ""
    exit_code: int | None = None
    stderr_output: str = ""

内容块类型

内部内容块 AssistantMessage.content:

类型	字段	描述
`TextBlock`	`text: str`	纯文本响应
`ThinkingBlock`	`thinking: str`	LLM内部推理
`ToolUseBlock`	`tool_use_id`, `name`, `input`	代理请求工具调用
`ToolResultBlock`	`tool_use_id`, `output`, `is_error`	工具执行结果

错误类型

流程失败会引发特定的异常，而不是产生错误事件：

异常	何时	属性
`CLINotFoundError`	`codebuddy` 不在PATH中	--
`AgentStartupError`	在产生输出之前，流程崩溃	`stderr`, `exit_code`
`AgentProcessError`	进程退出而不调用Complete/Block	`stderr`, `stdout_tail`, `exit_code`
`AgentExecutionError`	逻辑错误（身份验证失败、API错误）	`errors`, `subtype`

全部继承自 MCPAgentSDKError （继承自 Exception).

使用模式

自动重试验证

提供a validate_fn 以自动验证结果。如果验证失败，代理将收到反馈并重试最多 max_retries 时间。

def validate(result: str) -> tuple[bool, str]:
    if "success" in result.lower():
        return (True, "")
    return (False, "Result must indicate success. Please fix and try again.")

config = AgentRunConfig(
    prompt="Create and test a hello-world script",
    validate_fn=validate,
    max_retries=3,
)

回调

使用 on_complete 和 on_block 跑步结束时的副作用：

config = AgentRunConfig(
    prompt="Deploy the staging environment",
    on_complete=lambda result: notify_slack(f"✅ Deploy done: {result}"),
    on_block=lambda reason: page_oncall(f"🚧 Deploy blocked: {reason}"),
)

超时

设置最大持续时间。如果代理超过限制，则会自动阻止：

config = AgentRunConfig(
    prompt="Run the full test suite",
    timeout=120.0,  # 2 minutes
)

错误处理

包裹 run_agent() 在try/except中捕获进程失败：

from mcp_agent_sdk import AgentStartupError, AgentProcessError

try:
    async for event in sdk.run_agent(config):
        if isinstance(event, AgentResult):
            print(f"Result: {event.status}")
except AgentStartupError as e:
    print(f"CLI crashed on startup: {e}")
    print(f"stderr: {e.stderr}")
    print(f"exit code: {e.exit_code}")
except AgentProcessError as e:
    print(f"Agent died without completing: {e}")
    print(f"last output: {e.stdout_tail}")

并发代理

并行启动多个代理——每个代理都有自己的独立运行上下文：

async def run_all():
    sdk = MCPAgentSDK()
    await sdk.init()

    configs = [
        AgentRunConfig(prompt="Lint the codebase"),
        AgentRunConfig(prompt="Run unit tests"),
    ]

    async def _run(cfg):
        try:
            async for event in sdk.run_agent(cfg):
                if isinstance(event, AgentResult):
                    print(event.status)
        except (AgentStartupError, AgentProcessError) as e:
            print(f"Error: {e}")

    await asyncio.gather(*[_run(c) for c in configs])
    await sdk.shutdown()

自定义MCP服务器

通过以下方式将其他MCP服务器传递给代理子流程 mcp_servers。它们与内置 agent-controller 服务器（不能被覆盖）。

config = AgentRunConfig(
    prompt="Query our internal knowledge base and summarize results",
    mcp_servers={
        "knowledge-base": {
            "type": "http",
            "url": "http://localhost:9090/mcp",
        },
        "search-engine": {
            "command": "npx",
            "args": ["-y", "@anthropic/search-mcp-server"],
        },
    },
)

代理将可以访问所有配置的MCP服务器以及SDK内部的工具 agent-controller （完整/块工具）。

生命周期挂钩

使用钩子在关键生命周期点拦截和控制代理行为。钩子是异步回调，可以允许、阻止或修改代理操作。

支持的活动： PreToolUse, PostToolUse, UserPromptSubmit, Stop, SubagentStop, PreCompact

from mcp_agent_sdk import HookMatcher

async def block_dangerous_commands(hook_input, tool_use_id, context):
    """Block dangerous Bash commands before they execute."""
    input_data = hook_input.get("input", {})
    command = input_data.get("command", "")
    if any(d in command for d in ["rm -rf", "mkfs", "dd if="]):
        return {
            "continue_": False,
            "decision": "block",
            "reason": f"Blocked dangerous command: {command}",
        }
    return {"continue_": True}

config = AgentRunConfig(
    prompt="Clean up temp files in the current directory",
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",           # Only intercept Bash tool calls
                hooks=[block_dangerous_commands],
            )
        ],
    },
)

挂钩回拨签名

async def my_hook(
    hook_input: Any,            # Input data from CLI (tool name, args, etc.)
    tool_use_id: str | None,    # Tool use ID if applicable
    context: HookContext,       # {"signal": None}
) -> HookJSONOutput:
    return {"continue_": True}  # Allow the action

钩子返回值

字段	类型	描述
`continue_`	`bool`	是否继续执行(`True`)或阻止它(`False`)
`suppressOutput`	`bool`	是否抑制工具的输出
`stopReason`	`str`	停止原因（当 `continue_=False`)
`decision`	`str`	决策类型，例如。 `"block"`
`reason`	`str`	人类可读的决策原因

所有字段都是可选的。使用 continue_ （带尾随下划线）以避免Python关键字冲突——SDK将其映射到 continue 在协议中。

`HookMatcher`

@dataclass
class HookMatcher:
    matcher: str | None = None    # Tool name pattern to match (None = match all)
    hooks: list[HookCallback]     # List of async hook callbacks
    timeout: float | None = None  # Timeout for hook execution (seconds)

工具权限控制

使用 can_use_tool 在运行时以编程方式批准或拒绝工具调用。当CLI发送权限请求（需要 permission_mode 成为别的东西 "bypassPermissions").

from mcp_agent_sdk import (
    CanUseToolOptions, PermissionResultAllow, PermissionResultDeny,
)

ALLOWED_TOOLS = {"Read", "Glob", "Grep"}

async def my_permission_handler(
    tool_name: str,
    input_data: dict,
    options: CanUseToolOptions,
) -> PermissionResultAllow | PermissionResultDeny:
    if tool_name in ALLOWED_TOOLS:
        return PermissionResultAllow()
    return PermissionResultDeny(
        message=f"Tool '{tool_name}' is not in the allow-list",
    )

config = AgentRunConfig(
    prompt="Read and summarize pyproject.toml",
    can_use_tool=my_permission_handler,
    permission_mode="default",  # Required — bypass mode skips permission checks
)

要点：

can_use_tool=None （默认）使用 default_deny_can_use_tool，它拒绝所有工具调用，防止代理挂起未答复的权限请求。
PermissionResultAllow(updated_input={...}) 可以在执行之前修改工具的输入参数。
PermissionResultDeny(message="...", interrupt=True) 可以完全停止执行。
钩子(PreToolUse)先跑；如果钩子挡住了工具， can_use_tool 从未被召唤。
回调异常被捕获并自动转换为拒绝响应。

权限类型

类型	字段	描述
`CanUseToolOptions`	`tool_use_id`, `agent_id`	传递给回调的上下文
`PermissionResultAllow`	`behavior="allow"`, `updated_input`	允许工具调用，可选择修改输入
`PermissionResultDeny`	`behavior="deny"`, `message`, `interrupt`	拒绝工具调用并说明理由

运作原理

Your Code ──► MCPAgentSDK.run_agent(config)
                │
                ├─ Start MCP HTTP server (JSON-RPC 2.0)
                ├─ Register RunContext with unique agent_run_id
                ├─ Inject system prompt with Complete/Block tool instructions
                ├─ Launch codebuddy subprocess with MCP config
                ├─ Send hooks config via stdin control protocol (if configured)
                ├─ Start async stderr reader (deque buffer, last 100 lines)
                │
                │   ┌─────────────────────────────────────┐
                │   │  codebuddy agent runs task           │
                │   │  ├─ calls Complete(result) ──────────┼──► validate_fn() ──► retry or complete
                │   │  ├─ calls Block(reason)   ──────────┼──► on_block callback
                │   │  ├─ triggers hook event   ──────────┼──► hook callback ──► allow/block
                │   │  ├─ requests tool permission ────────┼──► can_use_tool() ──► allow/deny
                │   │  └─ crashes without calling either ──┼──► raise AgentProcessError(stderr)
                │   └─────────────────────────────────────┘
                │
                └─ Yield StreamEvent stream ──► your async for loop

发展

# Install dev dependencies
pip install -e ".[dev]"

# Run unit tests
pytest

# Run end-to-end tests (requires codebuddy in PATH)
pytest -m e2e

许可证

看许可证了解详情。

MCP Agent SDK

快速接入

详细介绍