AI Agent 协议全景：从 MCP 到 X42 的 6 大协议栈深度解析

# MCP Server 示例：GitHub 搜索工具
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx

server = Server("github-search")

@server.tool(
    name="search_repos",
    description="搜索 GitHub 仓库",
    inputSchema={
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "language": {"type": "string", "enum": ["python", "javascript", "go"]},
            "sort": {"type": "string", "enum": ["stars", "forks", "updated"]}
        },
        "required": ["query"]
    }
)
async def search_repos(query: str, language: str = None, sort: str = "stars"):
    """搜索 GitHub 仓库并返回结果"""
    async with httpx.AsyncClient() as client:
        params = {"q": query, "sort": sort}
        if language:
            params["language"] = language
        
        resp = await client.get(
            "https://api.github.com/search/repositories",
            params=params,
            headers={"Accept": "application/vnd.github.v3+json"}
        )
        data = resp.json()
    
    # 格式化结果
    results = []
    for repo in data["items"][:10]:
        results.append({
            "name": repo["full_name"],
            "stars": repo["stargazers_count"],
            "description": repo["description"],
            "url": repo["html_url"]
        })
    
    return TextContent(
        type="text",
        text=f"找到 {len(results)} 个仓库：\n" + 
             "\n".join([f"- {r['name']} ({r['stars']}⭐): {r['description']}" 
                        for r in results])
    )

# 启动服务器
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    
    async def main():
        async with stdio_server() as (read, write):
            await server.run(read, write)
    
    asyncio.run(main())

// Claude Desktop MCP 配置 (claude_desktop_config.json)
{
  "mcpServers": {
    "github": {
      "command": "python",
      "args": ["/path/to/github_server.py"],
      "env": {
        "GITHUB_TOKEN": "ghp_..."
      }
    },
    "database": {
      "command": "node",
      "args": ["/path/to/db_server.js"],
      "env": {
        "DATABASE_URL": "postgresql://..."
      }
    },
    "figma": {
      "url": "https://figma-mcp.example.com/sse",
      "headers": {
        "Authorization": "Bearer figd_..."
      }
    }
  }
}

{
  "name": "search_github",
  "description": "搜索 GitHub 仓库",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": { "type": "string" }
    },
    "required": ["query"]
  }
}

{
  "uri": "file:///path/to/file.txt",
  "name": "配置文件",
  "mimeType": "text/plain"
}

{
  "name": "code_review",
  "description": "代码审查流程",
  "arguments": [
    {
      "name": "repo_url",
      "description": "GitHub 仓库地址",
      "required": true
    }
  ]
}

# A2A Agent 示例：代码分析 Agent
from a2a import Agent, AgentCard, Task, Message
import httpx

class CodeAnalyzerAgent(Agent):
    def __init__(self):
        super().__init__(
            card=AgentCard(
                name="code-analyzer",
                description="分析代码质量、安全漏洞、性能瓶颈",
                capabilities={
                    "tasks": ["analyze_quality", "detect_vulnerabilities"],
                    "inputFormats": ["github_url"],
                    "outputFormats": ["json_report"]
                }
            )
        )
    
    async def handle_task(self, task: Task) -> dict:
        """处理代码分析任务"""
        if task.type == "analyze_quality":
            repo_url = task.input["repo_url"]
            
            # 使用 MCP 调用 GitHub API 下载代码
            code = await self.mcp_call("github", "download_repo", {"url": repo_url})
            
            # 使用 MCP 调用 SonarQube API 分析质量
            analysis = await self.mcp_call("sonarqube", "analyze", {"code": code})
            
            return {
                "status": "completed",
                "result": {
                    "quality_score": analysis["quality_gate"],
                    "issues": analysis["issues"],
                    "coverage": analysis["coverage"]
                }
            }
    
    async def handle_message(self, message: Message) -> Message:
        """处理来自其他 Agent 的消息"""
        if message.type == "task_delegation":
            task = Task(
                id=message.payload["task_id"],
                type=message.payload["task_type"],
                input=message.payload["input"]
            )
            result = await self.handle_task(task)
            
            return Message(
                from_agent=self.card.name,
                to_agent=message.from_agent,
                type="task_result",
                payload=result
            )

# 启动 Agent
if __name__ == "__main__":
    agent = CodeAnalyzerAgent()
    agent.run(host="0.0.0.0", port=8080)

{
  "name": "code-analyzer",
  "description": "分析代码质量、安全漏洞、性能瓶颈",
  "capabilities": {
    "tasks": ["analyze_quality", "detect_vulnerabilities", "profile_performance"],
    "inputFormats": ["github_url", "local_path"],
    "outputFormats": ["json_report", "pdf_report"]
  },
  "endpoint": "https://agents.example.com/code-analyzer"
}

{
  "id": "task-123",
  "type": "analyze_quality",
  "input": {
    "repo_url": "https://github.com/example/repo"
  },
  "status": "running",
  "context": {
    "parent_task": "task-122",
    "shared_data": {"user_email": "user@example.com"}
  }
}

{
  "from": "orchestrator-agent",
  "to": "code-analyzer",
  "type": "task_delegation",
  "payload": {
    "task_id": "task-123",
    "deadline": "2026-06-16T12:00:00Z",
    "priority": "high"
  }
}

// AG-UI + A2UI 组合示例：代码审查工具
import { Agent } from '@agent-sdk/core';
import { AGUIStream } from '@agent-sdk/ag-ui';
import { A2UIStore } from '@agent-sdk/a2-ui';

class CodeReviewAgent extends Agent {
  private agui: AGUIStream;
  private a2ui: A2UIStore;
  
  constructor() {
    super();
    this.agui = new AGUIStream();
    this.a2ui = new A2UIStore();
  }
  
  async reviewCode(repoUrl: string) {
    // AG-UI: 实时推送进度
    this.agui.push({ type: 'status', content: '正在下载代码...' });
    
    const code = await this.downloadCode(repoUrl);
    
    this.agui.push({ type: 'status', content: '正在分析代码...' });
    
    const issues = await this.analyzeCode(code);
    
    // A2UI: 持久化问题列表
    this.a2ui.update({
      issues: issues,
      resolvedIssues: [],
      comments: {}
    });
    
    // AG-UI: 请求用户确认
    this.agui.push({
      type: 'confirmation',
      content: `发现 ${issues.length} 个问题，是否生成修复建议？`,
      actions: ['yes', 'no', 'view_issues']
    });
    
    // 等待用户响应
    const userAction = await this.agui.waitForAction();
    
    if (userAction === 'yes') {
      // AG-UI: 流式生成修复建议
      for (const issue of issues) {
        this.agui.push({ type: 'token', content: `正在修复 ${issue.file}...` });
        const fix = await this.generateFix(issue);
        this.agui.push({ type: 'code_diff', file: issue.file, diff: fix });
        
        // A2UI: 更新修复状态
        this.a2ui.update({
          resolvedIssues: [...this.a2ui.state.resolvedIssues, issue.id]
        });
      }
    }
  }
}

// AG-UI 流式推送示例
const eventSource = new EventSource('/agent/stream');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  
  switch (data.type) {
    case 'token':
      // 实时显示生成的文本
      appendText(data.content);
      break;
    case 'tool_call':
      // 显示工具调用进度
      showToolProgress(data.tool, data.status);
      break;
    case 'confirmation':
      // 请求用户确认
      showConfirmationDialog(data.message, data.actions);
      break;
  }
};

// A2UI 状态同步示例
interface AgentState {
  tasks: Task[];
  currentTask: string;
  formData: Record<string, any>;
  approvals: Approval[];
}

// Agent 更新状态
agent.updateState({
  currentTask: 'task-123',
  formData: { name: 'Alice', email: 'alice@example.com' }
});

// UI 订阅状态变化
store.subscribe(() => {
  const state = store.getState();
  renderForm(state.formData);
  renderTaskList(state.tasks);
});

// 用户修改表单
store.dispatch({
  type: 'UPDATE_FORM',
  payload: { email: 'alice.new@example.com' }
});

// Agent 收到状态变化通知
agent.onStateChange((newState) => {
  console.log('用户修改了表单:', newState.formData);
});

# AP2 实战：部署和管理 Agent

# 1. 部署 Agent
ap2 apply -f - <<EOF
apiVersion: ap2/v1
kind: Agent
metadata:
  name: code-reviewer
  namespace: engineering
spec:
  image: agents/code-reviewer:v1.2.3
  replicas: 3
  capabilities:
    - code_review
    - vulnerability_detection
  resources:
    requests:
      cpu: "2"
      memory: "4Gi"
  scaling:
    minReplicas: 2
    maxReplicas: 10
    targetCPU: 70%
EOF

# 2. 查看 Agent 状态
ap2 get agents -n engineering
NAME            READY   STATUS    RESTARTS   AGE
code-reviewer   3/3     Running   0          5m

# 3. 查看 Agent 日志
ap2 logs agent/code-reviewer --tail=50

# 4. 升级 Agent
ap2 set image agent/code-reviewer agents/code-reviewer:v1.3.0

# 5. 回滚
ap2 rollout undo agent/code-reviewer

# 6. 查看 Agent 指标
ap2 top agent -n engineering
NAME            CPU     MEMORY   REQUESTS/s   LATENCY
code-reviewer   2.3/4   5.2/8G   120          450ms

# X42 实战：跨域 Agent 调用
from x42 import X42Client, DID, VerifiableCredential
import asyncio

async def cross_org_agent_call():
    # 1. 加载身份
    my_did = DID.load("did:x42:agent:data-analyzer:company-b")
    my_key = my_did.load_key("key-1")
    
    # 2. 获取目标 Agent 的 DID 文档
    target_did = await DID.resolve("did:x42:agent:code-reviewer:company-a")
    target_endpoint = target_did.service[0].serviceEndpoint
    
    # 3. 创建能力凭证（证明我有权调用）
    credential = VerifiableCredential(
        issuer=my_did,
        subject=my_did,
        capability={
            "invoke": ["code_review"],
            "access": ["github:company-b/*"],
            "quota": {"requests_per_day": 1000}
        }
    )
    await credential.sign(my_key)
    
    # 4. 调用目标 Agent
    client = X42Client()
    response = await client.invoke(
        endpoint=target_endpoint,
        action="code_review",
        input={"repo": "github.com/company-b/backend"},
        credential=credential
    )
    
    print(f"发现 {response['issues']} 个问题")

asyncio.run(cross_org_agent_call())

# ap2-agent.yaml
apiVersion: ap2/v1
kind: Agent
metadata:
  name: code-reviewer
  namespace: engineering
  labels:
    team: backend
    tier: production
spec:
  image: agents/code-reviewer:v1.2.3
  replicas: 3
  capabilities:
    - code_review
    - vulnerability_detection
  resources:
    requests:
      cpu: "2"
      memory: "4Gi"
    limits:
      cpu: "4"
      memory: "8Gi"
  healthCheck:
    path: /health
    interval: 30s
  scaling:
    minReplicas: 2
    maxReplicas: 10
    targetCPU: 70%

# 部署 Agent
ap2 apply -f ap2-agent.yaml

# 查看 Agent 状态
ap2 get agents -n engineering
NAME            READY   STATUS    RESTARTS   AGE
code-reviewer   3/3     Running   0          5m

# 升级 Agent
ap2 set image agent/code-reviewer agents/code-reviewer:v1.3.0

# 回滚
ap2 rollout undo agent/code-reviewer

# 查看 Agent 日志
ap2 logs agent/code-reviewer --tail=100

# 查看 Agent 指标
ap2 top agent -n engineering
NAME            CPU     MEMORY   REQUESTS/s   LATENCY
code-reviewer   2.3/4   5.2/8G   120          450ms
data-analyzer   1.8/2   3.1/4G   85           620ms

{
  "id": "did:x42:agent:code-reviewer:company-a",
  "verificationMethod": [{
    "id": "#key-1",
    "type": "Ed25519VerificationKey2020",
    "publicKeyMultibase": "z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK"
  }],
  "service": [{
    "id": "#agent-endpoint",
    "type": "AgentService",
    "serviceEndpoint": "https://agents.company-a.com/code-reviewer"
  }]
}

{
  "@context": "https://www.w3.org/2018/credentials/v1",
  "type": ["VerifiableCredential", "AgentCapabilityCredential"],
  "issuer": "did:x42:org:company-a",
  "issuanceDate": "2026-06-16T00:00:00Z",
  "expirationDate": "2027-06-16T00:00:00Z",
  "credentialSubject": {
    "id": "did:x42:agent:code-reviewer:company-a",
    "capability": {
      "invoke": ["code_review", "vulnerability_detection"],
      "access": ["github:company-a/*", "jira:company-a/*"],
      "quota": {
        "requests_per_day": 10000,
        "max_payload_size": "10MB"
      }
    }
  },
  "proof": {
    "type": "Ed25519Signature2020",
    "created": "2026-06-16T00:00:00Z",
    "verificationMethod": "did:x42:org:company-a#key-1",
    "proofValue": "z58DAdFfa9SkqZMVPxAQpic7ndTn3nnz..."
  }
}

{
  "timestamp": "2026-06-16T10:30:00Z",
  "caller": "did:x42:agent:data-analyzer:company-b",
  "callee": "did:x42:agent:code-reviewer:company-a",
  "action": "code_review",
  "input": {"repo": "github.com/company-a/backend"},
  "output": {"issues": 15, "vulnerabilities": 2},
  "duration_ms": 4500,
  "authorized_by": "did:x42:policy:cross-org-access"
}