a2a_auto_wrapper

将 CrewAI Agent 一行命令转为 A2A 服务 — 开源替代 CrewAI AMP 的本地开发方案。

基于 AutoA2A (NapthaAI) 和 Google A2A Protocol。

安装

pip install a2a_auto_wrapper

同时需要安装 CrewAI：

pip install crewai

快速开始

服务端 — 一行启动 CrewAI A2A 服务

# crewai_a2a_server.py
from crewai import Agent, LLM
from crewai.a2a import A2AServerConfig
from a2a_auto_wrapper import serve_crewai_agent

analyst = Agent(
    role="数据分析师",
    goal="对数据进行深度分析并生成结构化报告",
    backstory="你是资深数据分析师，擅长数据洞察和趋势预测。",
    llm=LLM(model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="..."),
    a2a=A2AServerConfig(url="http://localhost:2025"),
)

if __name__ == "__main__":
    serve_crewai_agent(analyst, port=2025)

python crewai_a2a_server.py
# ✅ 服务运行在 http://localhost:2025

客户端 — 调用 A2A 服务

import asyncio
from a2a_auto_wrapper import A2AClient

async def main():
    client = A2AClient("http://localhost:2025/")
    result = await client.send_message("分析销售数据：1月100万 2月120万 3月90万")
    print(result)

asyncio.run(main())

DeepAgents 调用 CrewAI（A2A 互调）

# deepagent_a2a.py
from a2a_auto_wrapper import A2AClient
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from deepagents import create_deep_agent

crewai_client = A2AClient(base_url="http://localhost:2025/")

@tool
async def call_crewai_analyst(query: str) -> str:
    """需要数据分析时调用 CrewAI 分析师。"""
    return await crewai_client.send_message(query)

agent = create_deep_agent(
    model=ChatOpenAI(model="deepseek-chat", ...),
    tools=[call_crewai_analyst],
)
graph = agent  # langgraph dev 加载入口

API 参考

serve_crewai_agent(agent, port=2025)

启动 CrewAI Agent 为 A2A HTTP 服务。

A2AClient

A2A 客户端，调用远程 A2A 服务。

client = A2AClient(base_url: str, timeout: int = 120)
result = await client.send_message(text: str) -> str

通信流程

用户 → langgraph_sdk → DeepAgents (:2024) → A2A → CrewAI (:2025)

依赖

• Python >= 3.12
• crewai (安装本包后需单独安装)
• httpx, starlette, uvicorn, pydantic

License

MIT