# MCP Server 示例：暴露工具给Agent使用
from mcp.server import Server

server = Server("research-tools")

@server.tool("search_database")
async def search_database(query: str, limit: int = 10) -> dict:
    """搜索研究数据库"""
    results = await db.search(query, limit=limit)
    return {"results": results, "count": len(results)}

@server.tool("generate_report")
async def generate_report(data: dict, format: str = "pdf") -> dict:
    """生成分析报告"""
    report = await report_engine.create(data, format=format)
    return {"url": report.url, "pages": report.page_count}

{
  "@type": "AgentCard",
  "name": "Research Analyst Agent",
  "description": "深度研究分析Agent，擅长市场调研和竞品分析",
  "url": "https://agents.example.com/research-analyst",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "name": "market_research",
      "description": "执行市场调研并生成结构化报告"
    },
    {
      "name": "competitor_analysis",
      "description": "分析竞争对手的产品策略和技术路线"
    }
  ],
  "authentication": {
    "schemes": ["Bearer Token"]
  }
}

┌─────────────────────────────────────┐
│         AG-UI (用户交互层)            │
│    流式渲染 / 状态同步 / 人机协作      │
├─────────────────────────────────────┤
│         A2A (Agent协作层)             │
│  Agent发现 / 任务委托 / 结果回传      │
├─────────────────────────────────────┤
│         MCP (工具接入层)              │
│   工具调用 / 数据源 / API集成         │
├─────────────────────────────────────┤
│         LLM (推理引擎层)             │
│   GPT-4.1 / Claude / Gemini        │
└─────────────────────────────────────┘

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

# 定义专业Agent
research_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[search_web, read_document],
    name="researcher",
    prompt="你是研究专家，负责收集和分析信息。"
)

writer_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[write_file, format_text],
    name="writer",
    prompt="你是写作专家，负责将研究结果转化为高质量文章。"
)

# 创建Supervisor编排
supervisor = create_supervisor(
    model="openai:gpt-4.1",
    agents=[research_agent, writer_agent],
    prompt="""你是一个任务编排器。根据用户需求：
    - 研究类任务交给 researcher
    - 写作类任务交给 writer
    - 复杂任务先让 researcher 调研，再交给 writer 写作"""
)

# 编译并运行
app = supervisor.compile()
result = await app.ainvoke({
    "messages": [{"role": "user", "content": "调研2026年AI Agent市场并写一份报告"}]
})

from langgraph_swarm import create_swarm, swarm_agent

@swarm_agent
async def triage_agent(state):
    """分类Agent：判断用户意图"""
    # Agent自己决定下一步：直接回复，还是handoff
    return {
        "response": "...",
        "next_agent": "billing_agent"  # 自主决定转交
    }

@swarm_agent
async def billing_agent(state):
    """账单Agent：处理计费问题"""
    return {
        "response": "已为您查询到...",
        "next_agent": None  # 任务完成，不转交
    }

swarm = create_swarm(
    agents=[triage_agent, billing_agent, tech_support_agent],
    default_agent="triage_agent"
)

from crewai import Agent, Task, Crew, Process

# 定义Worker Agent
researcher = Agent(
    role="高级研究员",
    goal="深入调研技术主题，收集全面信息",
    backstory="10年技术研究经验的资深分析师",
    tools=[search_tool, scrape_tool],
    llm="gpt-4.1"
)

analyst = Agent(
    role="数据分析师",
    goal="从研究数据中提取关键洞察",
    backstory="擅长数据可视化和趋势分析的数据科学家",
    tools=[python_tool, chart_tool],
    llm="gpt-4.1"
)

writer = Agent(
    role="技术写作专家",
    goal="将分析结果转化为专业报告",
    backstory="曾就职于顶级科技媒体的技术编辑",
    llm="gpt-4.1"
)

# 创建层级Crew（自动生成Manager）
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[...],  # Manager会根据需要创建和分配任务
    process=Process.hierarchical,
    manager_llm="gpt-4.1",
    verbose=True
)

result = crew.kickoff(inputs={"topic": "2026年多智能体系统技术趋势"})

# MAD辩论框架核心逻辑（简化实现）
class MultiAgentDebate:
    def __init__(self, agents: list[Agent], rounds: int = 3):
        self.agents = agents
        self.rounds = rounds
        self.judge = Agent(role="judge", model="gpt-4.1")

    async def debate(self, question: str) -> dict:
        arguments = {i: "" for i in range(len(self.agents))}

        for round_num in range(self.rounds):
            for i, agent in enumerate(self.agents):
                # 构建辩论上下文：看到所有其他Agent的论点
                context = self._build_context(arguments, exclude=i)
                arguments[i] = await agent.generate(
                    f"问题: {question}\n"
                    f"其他Agent的观点:\n{context}\n"
                    f"请{'提出你的初始论点' if round_num == 0 else '回应质疑并完善论点'}"
                )

        # Judge汇总最终结论
        final = await self.judge.generate(
            f"问题: {question}\n"
            f"各方论点:\n{self._summarize(arguments)}\n"
            f"请综合各方观点，给出最终结论。"
        )
        return {"conclusion": final, "arguments": arguments}

from agents import Agent, Runner, handoff

# 定义专业Agent
spanish_agent = Agent(
    name="spanish_agent",
    instructions="你用西班牙语回答问题。",
    model="gpt-4.1-mini"
)

french_agent = Agent(
    name="french_agent",
    instructions="你用法语回答问题。",
    model="gpt-4.1-mini"
)

# 主Agent通过handoff机制转移控制权
triage_agent = Agent(
    name="triage_agent",
    instructions="""根据用户语言选择合适的Agent。
    西班牙语→spanish_agent，法语→french_agent""",
    model="gpt-4.1",
    handoffs=[
        handoff(spanish_agent),
        handoff(french_agent)
    ]
)

result = await Runner.run(triage_agent, "Hola, ¿cómo estás?")

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 配置持久化检查点
checkpointer = MemorySaver()  # 生产环境替换为PostgreSQL/Redis

# 编译图时启用检查点
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # 人工审核节点前暂停
)

# 使用thread_id追踪会话状态
config = {"configurable": {"thread_id": "session-001"}}
result = await app.ainvoke(input_data, config=config)

# 可以随时恢复执行
resumed = await app.ainvoke(None, config=config)  # 从检查点恢复

# A2A客户端：发现并调用远程Agent
import httpx

async def discover_agent(agent_card_url: str) -> dict:
    """发现远程Agent并获取其能力描述"""
    async with httpx.AsyncClient() as client:
        response = await client.get(agent_card_url)
        return response.json()  # 返回Agent Card

async def delegate_to_agent(agent_url: str, task: str) -> dict:
    """通过A2A协议向远程Agent委托任务"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{agent_url}/tasks",
            json={
                "task": {
                    "id": "task-001",
                    "message": {
                        "role": "user",
                        "parts": [{"type": "text", "text": task}]
                    }
                }
            }
        )
        return response.json()

# 使用示例
agent_card = await discover_agent(
    "https://agents.partner.com/analytics/.well-known/agent.json"
)
result = await delegate_to_agent(
    agent_card["url"],
    "分析过去30天的用户行为数据，识别流失风险用户"
)