{
  "agent_card": {
    "name": "billing-agent",
    "description": "处理账单查询和退款",
    "capabilities": ["billing_query", "refund_processing"],
    "endpoint": "https://billing.example.com/a2a",
    "authentication": {"scheme": "oauth2"}
  }
}

# A2A 协议任务委托示例
from a2a.client import A2AClient

client = A2AClient()
# 发现远程Agent
agent = await client.discover("https://partner.com/.well-known/agent.json")

# 委托任务
task = await agent.send_task({
    "intent": "analyze_financial_report",
    "payload": {"report_id": "Q1-2026", "metrics": ["revenue", "margin"]}
})

# 流式接收结果
async for event in task.stream():
    print(f"进度: {event.progress}% - {event.status}")

# MCP 2026 - Agent作为工具暴露给其他Agent
from mcp.server import MCPServer

server = MCPServer("research-agent")

@server.tool("deep_research")
async def deep_research(topic: str, depth: int = 3):
    """暴露研究Agent能力为MCP工具"""
    # 内部可能编排多个子Agent
    planner = ResearchPlanner()
    executor = ResearchExecutor()
    synthesizer = Synthesizer()

    plan = await planner.create_plan(topic, depth)
    results = await executor.execute_parallel(plan)
    return await synthesizer.synthesize(results)

# AG2 编排模式 - Round Robin
import ag2 as autogen
from ag2 import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

llm_config = {"model": "gpt-4.1", "api_key": "your-key"}

researcher = AssistantAgent("researcher", llm_config=llm_config,
    system_message="你负责信息收集和研究分析")
writer = AssistantAgent("writer", llm_config=llm_config,
    system_message="你负责将研究结果转化为文章")
reviewer = AssistantAgent("reviewer", llm_config=llm_config,
    system_message="你负责审校和优化文章质量")

# Round Robin: 固定轮次顺序
groupchat = GroupChat(
    agents=[researcher, writer, reviewer],
    messages=[],
    max_round=6,
    speaker_selection_method="round_robin"  # 也可选 "auto", "random", "custom"
)

manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
user = UserProxyAgent("user", human_input_mode="NEVER")
user.initiate_chat(manager, message="分析2026年多智能体系统的市场趋势")

# OpenAI Agents SDK - Handoff接力
from agents import Agent, Runner, handoff

# 定义三个专业Agent
triage_agent = Agent(
    name="triage",
    instructions="分析用户问题并决定转交给哪个专业Agent处理",
    handoffs=[
        handoff(agent_name="technical", description="技术问题、Bug报告、API使用"),
        handoff(agent_name="billing", description="账单、退款、订阅管理"),
    ]
)

technical_agent = Agent(
    name="technical",
    instructions="处理技术支持问题。如果问题涉及退款，交接给billing agent。",
    handoffs=[handoff(agent_name="billing")],
    tools=[search_docs, run_diagnostics]
)

billing_agent = Agent(
    name="billing",
    instructions="处理账单相关问题",
    tools=[lookup_charges, process_refund],
    handoffs=[handoff(agent_name="technical")]
)

# 运行 - 自动处理交接
result = await Runner.run(triage_agent, messages=[
    {"role": "user", "content": "我的订阅被扣了两次费用，而且API还报错了"}
])
# 内部流程: triage -> billing (处理退款) -> technical (处理API错误)

// Microsoft Agent Framework - .NET 10 分层Process
using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Process;

var builder = Kernel.CreateBuilder();
builder.AddAzureOpenAIChatCompletion(/* config */);
var kernel = builder.Build();

// 定义三层Process
var process = ProcessBuilder
    .WithName("EnterpriseAnalysis")
    .AddStep<PlannerAgent>("planner",    // 顶层：战略规划
        description: "分解业务目标为可执行任务")
    .AddStep<CoordinatorAgent>("coord",  // 中层：任务协调
        description: "分配任务给专业Agent并监控进度")
    .AddStep<DataAgent>("data",          // 底层：数据收集
        description: "从数据库和API收集数据")
    .AddStep<AnalysisAgent>("analyst",   // 底层：数据分析
        description: "执行统计分析和建模")
    .AddStep<ReportAgent>("report",      // 底层：报告生成
        description: "生成可视化报告")
    // 定义层级关系
    .OnEvent("plan_ready").SendTo("coord")
    .OnEvent("need_data").SendTo("data")
    .OnEvent("need_analysis").SendTo("analyst")
    .OnEvent("analysis_done").SendTo("report")
    .Build();

await process.StartAsync(new { objective = "Q1市场份额分析" });

# 黑板模式实现 - 使用Redis作为共享存储
import json, redis
from datetime import datetime

class Blackboard:
    def __init__(self):
        self.store = redis.Redis(host='localhost', port=6379, db=0)

    def write(self, agent_id: str, section: str, data: dict):
        entry = {"agent": agent_id, "data": data, "timestamp": datetime.now().isoformat()}
        self.store.hset("blackboard", section, json.dumps(entry))
        self.store.publish("blackboard_update", json.dumps({"section": section, "agent": agent_id}))

    def read(self, section: str = None) -> dict:
        if section:
            return json.loads(self.store.hget("blackboard", section))
        return {k: json.loads(v) for k, v in self.store.hgetall("blackboard").items()}

# 医疗诊断多Agent系统
class DiagnosticAgent:
    def __init__(self, name, specialty, blackboard):
        self.name = name
        self.specialty = specialty
        self.bb = blackboard

    async def analyze(self, patient_data):
        # 读取黑板上的所有已有分析
        current_state = self.bb.read()
        # 基于自身专长进行分析
        findings = await self.llm_analyze(patient_data, current_state)
        # 写入自己的分析结果到黑板
        self.bb.write(self.name, self.specialty, findings)

# 使用示例
bb = Blackboard()
agents = [
    DiagnosticAgent("radiologist", "imaging", bb),
    DiagnosticAgent("pathologist", "lab_results", bb),
    DiagnosticAgent("clinician", "symptoms", bb),
]

# AG2 辩论模式
from ag2 import AssistantAgent, GroupChat, GroupChatManager

proponent = AssistantAgent("proponent", llm_config=config,
    system_message="""你负责支持给定的方案。
    提供有力的论据、数据和案例支持。
    每轮提出1-2个新论点。""")

opponent = AssistantAgent("opponent", llm_config=config,
    system_message="""你负责质疑和反对给定的方案。
    指出风险、成本和潜在问题。
    提出替代方案。""")

judge = AssistantAgent("judge", llm_config=config,
    system_message="""你是中立的裁判。
    评估双方的论据，指出逻辑漏洞。
    3轮辩论后给出最终裁决和改进建议。""")

def debate_selector(last_speaker, messages):
    """自定义发言选择器 - 确保辩论有序进行"""
    if last_speaker == proponent: return opponent
    if last_speaker == opponent: return judge if len(messages) > 10 else proponent
    return proponent  # 裁判发言后由支持方继续

groupchat = GroupChat(
    agents=[proponent, opponent, judge],
    speaker_selection_method=debate_selector,
    max_round=12
)

# Strands Agents - 模型驱动多Agent
from strands import Agent, AgentCore, tool

@tool
def search_database(query: str) -> str:
    """搜索产品数据库"""
    return db.search(query)

@tool  
def analyze_sentiment(text: str) -> dict:
    """分析客户情感"""
    return {"sentiment": "positive", "score": 0.85}

# 定义协作Agent
research_agent = Agent(
    name="researcher",
    description="负责市场研究和数据收集",
    tools=[search_database, web_search]
)

analysis_agent = Agent(
    name="analyst", 
    description="负责数据分析和洞察生成",
    tools=[analyze_sentiment, statistical_analysis]
)

# 主Agent - 由模型自动决定何时委派
orchestrator = Agent(
    model="claude-sonnet-4-20250514",
    tools=[research_agent, analysis_agent],  # Agent作为工具使用
    system_prompt="你是一个市场分析协调者。根据任务自动决定委派给哪个Agent。"
)

# 一行代码运行 - 模型自动编排
result = orchestrator("分析我们Q1产品评论的情感趋势")

# 成本控制策略示例
from agents import Agent, Runner, RunConfig

config = RunConfig(
    max_turns=8,              # 限制最大轮次
    max_tokens_per_turn=2000, # 每轮限制token
    model_tier="auto",        # 自动选择模型层级
    # 简单路由用haiku，复杂推理用sonnet
    model_routing={
        "triage": "claude-haiku-4-5-20251001",
        "analysis": "claude-sonnet-4-20250514",
        "review": "claude-sonnet-4-20250514"
    }
)