DeepAgents + MCP + A2A + Skills：构建可组合的智能体系统

from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model

model = init_chat_model("anthropic:claude-sonnet-4-20250514")

agent = create_deep_agent(
    model=model,
    tools=[],                          # 工具通过 MCP 注入，见下节
    system_prompt=(
        "你是技术负责人。先规划，再委派子智能体，最后汇总。"
        "关键中间产物写入文件，避免上下文溢出。"
    ),
    subagents=[
        {"name": "researcher", "description": "检索资料，输出事实要点",
         "prompt": "只输出可验证事实，附来源。"},
        {"name": "coder", "description": "实现代码并通过测试",
         "prompt": "先写测试，再实现，运行 pytest 直到通过。"},
    ],
)

.claude/skills/security-review/
└── SKILL.md

---
name: security-review
description: 审查代码的安全问题，覆盖注入、鉴权、密钥泄露
---

# 安全审查 Skill

## 触发条件
当任务涉及鉴权、支付、外部输入处理时激活。

## 执行步骤
1. 用 Grep 搜索危险模式：eval、exec、innerHTML、SQL 拼接
2. 检查所有外部输入是否校验
3. 检查密钥是否硬编码
4. 输出问题清单，按 CVSS 分级

## 输出格式
| 严重度 | 文件:行 | 问题 | 修复建议 |

import pathlib

def load_skills(skill_dir=".claude/skills") -> str:
    parts = []
    for p in pathlib.Path(skill_dir).glob("*/SKILL.md"):
        parts.append(p.read_text(encoding="utf-8"))
    return "\n\n---\n\n".join(parts)

agent = create_deep_agent(
    model=model,
    system_prompt="你是技术负责人。\n\n可用技能：\n" + load_skills(),
)

from langchain_mcp_adapters.client import MultiServerMCPClient

mcp = MultiServerMCPClient({
    "postgres": {
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-postgres",
                 "postgresql://readonly@localhost/app"],
    },
    "github": {
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": {"GITHUB_TOKEN": "${GITHUB_TOKEN}"},
    },
})

tools = await mcp.get_tools()      # 自动发现所有工具
agent = create_deep_agent(model=model, tools=tools)

from a2a.server import A2AServer, AgentCard, Task
import httpx

card = AgentCard(
    name="security-auditor",
    description="对代码仓库做安全审计",
    url="http://auditor.internal:8080",
    capabilities={"streaming": False},
)

server = A2AServer(card)

@server.on_task
async def handle(task: Task) -> dict:
    repo = task.input["repo"]
    # 调用本地 DeepAgent 执行审计
    result = await agent.ainvoke({"messages":
        [{"role": "user", "content": f"审计仓库 {repo} 的安全问题"}]})
    return {"report": result["messages"][-1].content}

# 调用方：从 Agent Card 发现并委派
async def delegate(repo: str):
    card = (await httpx.get(
        "http://auditor.internal/.well-known/agent.json")).json()
    r = await httpx.post(f"{card['url']}/tasks", json={
        "input": {"repo": repo}
    }, timeout=300)
    return r.json()

async def audit_pipeline(repo: str):
    # 1. 通过 A2A 委派给安全审计 Agent
    report = await delegate(repo)
    # 2. 本地 DeepAgent 结合 Skill 做修复规划
    plan = await agent.ainvoke({"messages": [{"role": "user",
        "content": f"基于审计报告制定修复计划：\n{report['report']}"}]})
    # 3. 用 MCP 工具创建 GitHub Issue
    gh = next(t for t in tools if t.name == "create_issue")
    await gh.ainvoke({"title": "安全修复计划",
                      "body": plan["messages"][-1].content})
    return plan