04 - ACP SDK 与实现方案

1. 官方 SDK 概览

1.1 SDK 生态系统

ACP 提供了多语言的官方 SDK 支持，使开发者能够快速构建符合协议的 Agent[1]：

flowchart TB
    subgraph "ACP SDK 生态"
        Core[ACP Core Protocol]
        
        subgraph "官方 SDK"
            Python[Python SDK<br/>acp-sdk]
            TypeScript[TypeScript SDK<br/>acp-sdk]
        end
        
        subgraph "社区 SDK"
            Go[Go SDK<br/>coder/acp-go-sdk]
            Java[Java SDK<br/>solonlab/acp-java-sdk]
            Kotlin[Kotlin SDK<br/>agentclientprotocol]
        end
    end
    
    Core --> Python
    Core --> TypeScript
    Core -.-> Go
    Core -.-> Java
    Core -.-> Kotlin

1.2 SDK 功能对比

xychart-beta
    title "各 SDK 功能完整性对比"
    x-axis ["Python", "TypeScript", "Go", "Java", "Kotlin"]
    y-axis "功能完整度 (%)"
    bar [95, 85, 70, 40, 60]

注：⚠️ 表示部分支持，❌ 表示不支持

2. Python SDK 详解

2.1 安装与配置

Python SDK 可通过 pip 或 uv 安装[2]：

# 使用 pip
pip install acp-sdk

# 使用 uv（推荐）
uv add acp-sdk

环境要求：

• Python >= 3.11
• 异步运行时（asyncio）

2.2 快速入门示例

2.2.1 创建简单的 Echo Agent

import asyncio
from collections.abc import AsyncGenerator

from acp_sdk.models import Message
from acp_sdk.server import Context, RunYield, RunYieldResume, Server

server = Server()

@server.agent()
async def echo(
    input: list[Message], context: Context
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """Echoes everything"""
    for message in input:
        await asyncio.sleep(0.5)
        yield {"thought": "I should echo everything"}
        await asyncio.sleep(0.5)
        yield message

if __name__ == "__main__":
    server.run()

启动 Agent 服务：

uv run agent.py
# 服务运行在 http://localhost:8000

2.2.2 创建 ACP 客户端

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def example() -> None:
    async with Client(base_url="http://localhost:8000") as client:
        # 同步调用
        run = await client.run_sync(
            agent="echo",
            input=[
                Message(
                    role="user",
                    parts=[MessagePart(
                        content="Hello from ACP!", 
                        content_type="text/plain"
                    )]
                )
            ],
        )
        print(run.output)

if __name__ == "__main__":
    asyncio.run(example())

2.3 高级功能实现

2.3.1 集成 LangChain Agent

from typing import Annotated
from typing_extensions import TypedDict
from dotenv import load_dotenv

# ACP SDK
from acp_sdk.models import Message, MessagePart
from acp_sdk.server import RunYield, RunYieldResume, Server
from collections.abc import AsyncGenerator

# LangChain
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic

load_dotenv()

class State(TypedDict):
    messages: Annotated[list, add_messages]

# 设置 AI 模型
llm = ChatAnthropic(
    model="claude-3-5-sonnet-latest",
    api_key=os.environ.get("ANTHROPIC_API_KEY")
)

server = Server()

@server.agent()
async def chatbot(
    messages: list[Message]
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """A simple chatbot with memory"""
    # 转换 ACP 消息格式为 LangChain 格式
    query = " ".join(
        part.content
        for m in messages
        for part in m.parts
    )
    
    # 调用 LLM
    llm_response = llm.invoke(query)
    
    # 转换回 ACP 格式
    assistant_message = Message(
        parts=[MessagePart(content=llm_response.content)]
    )
    
    yield {"messages": [assistant_message]}

server.run()

2.3.2 流式响应实现

@server.agent()
async def streaming_agent(
    input: list[Message], context: Context
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """Agent with streaming support"""
    query = extract_text(input)
    
    # 模拟流式生成
    tokens = ["Hello", ",", " how", " can", " I", " help", " you", "?"]
    
    for token in tokens:
        await asyncio.sleep(0.2)
        yield {
            "parts": [MessagePart(
                content=token, 
                content_type="text/plain"
            )]
        }
    
    # 发送完成状态
    yield {"status": "completed"}

2.3.3 会话状态管理

@server.agent()
async def stateful_agent(
    input: list[Message], context: Context
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """Agent with session memory"""
    session_id = context.session_id
    
    # 获取会话历史
    history = await context.get_session_history(session_id)
    
    # 结合历史和新输入生成响应
    full_context = history + input
    response = await generate_response(full_context)
    
    # 更新会话历史
    await context.update_session(session_id, input + [response])
    
    yield {"messages": [response]}

2.4 高可用部署

from acp_sdk.server import Server
from acp_sdk.storage import RedisStorage

# 使用 Redis 作为共享存储
server = Server(
    storage=RedisStorage(
        host="redis-cluster.example.com",
        port=6379,
        db=0,
        password=os.environ.get("REDIS_PASSWORD")
    )
)

# 启用健康检查
server.enable_health_check(path="/health")

# 配置限流
server.configure_rate_limit(
    requests_per_minute=100,
    burst_size=10
)

server.run(
    host="0.0.0.0",
    port=8000,
    workers=4  # 多 worker 模式
)

3. TypeScript SDK 详解

3.1 安装

npm install acp-sdk
# 或
yarn add acp-sdk
# 或
pnpm add acp-sdk

3.2 客户端示例

import { Client, Message, MessagePart } from 'acp-sdk';

async function main() {
  const client = new Client({
    baseUrl: 'http://localhost:8000'
  });

  try {
    // 同步调用
    const run = await client.runSync({
      agent: 'echo',
      input: [
        new Message({
          role: 'user',
          parts: [
            new MessagePart({
              content: 'Hello TypeScript!',
              contentType: 'text/plain'
            })
          ]
        })
      ]
    });

    console.log('Response:', run.output);

    // 流式调用
    const stream = await client.runStream({
      agent: 'chatbot',
      input: [/* ... */]
    });

    for await (const chunk of stream) {
      process.stdout.write(chunk.content);
    }

  } finally {
    await client.close();
  }
}

main().catch(console.error);

3.3 与前端框架集成

// React Hook 示例
import { useState, useCallback } from 'react';
import { Client } from 'acp-sdk';

const useACPAgent = (agentName: string) => {
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState<Error | null>(null);
  
  const client = new Client({ 
    baseUrl: process.env.REACT_APP_ACP_URL 
  });

  const sendMessage = useCallback(async (message: string) => {
    setIsLoading(true);
    setError(null);
    
    try {
      const response = await client.runSync({
        agent: agentName,
        input: [{
          role: 'user',
          parts: [{
            content: message,
            contentType: 'text/plain'
          }]
        }]
      });
      
      return response.output;
    } catch (err) {
      setError(err as Error);
      throw err;
    } finally {
      setIsLoading(false);
    }
  }, [agentName]);

  return { sendMessage, isLoading, error };
};

export default useACPAgent;

4. 无需 SDK 的 HTTP 调用

ACP 的一个核心优势是无需 SDK 即可调用[3]：

4.1 使用 cURL

# 列出可用 Agent
curl http://localhost:8000/agents

# 同步调用 Agent
curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
    "agent_name": "echo",
    "input": [
      {
        "role": "user",
        "parts": [
          {
            "content": "Hello via cURL!",
            "content_type": "text/plain"
          }
        ]
      }
    ]
  }'

# 流式调用
curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "agent_name": "streaming-agent",
    "mode": "streaming",
    "input": [...]
  }'

4.2 使用 Postman

4.3 使用浏览器 Fetch

const callACPAgent = async () => {
  const response = await fetch('http://localhost:8000/runs', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      agent_name: 'echo',
      input: [{
        role: 'user',
        parts: [{
          content: 'Hello from browser!',
          content_type: 'text/plain'
        }]
      }]
    })
  });

  const result = await response.json();
  console.log(result);
};

5. 最佳实践

5.1 架构设计原则

flowchart TB
    subgraph "ACP 最佳实践架构"
        LB[负载均衡器]
        
        subgraph "ACP 服务层"
            S1[Server 1]
            S2[Server 2]
            S3[Server 3]
        end
        
        subgraph "存储层"
            Redis[(Redis)]
            PG[(PostgreSQL)]
        end
        
        subgraph "Agent 层"
            A1[Agent A]
            A2[Agent B]
            A3[Agent C]
        end
    end
    
    LB --> S1
    LB --> S2
    LB --> S3
    
    S1 --> Redis
    S2 --> Redis
    S3 --> Redis
    
    S1 --> PG
    
    S1 --> A1
    S2 --> A2
    S3 --> A3

5.2 性能优化建议

5.3 错误处理模式

from acp_sdk.client import Client
from acp_sdk.exceptions import AgentNotFoundError, AgentError, TimeoutError

async def robust_call():
    client = Client(base_url="http://localhost:8000")
    
    try:
        result = await client.run_sync(agent="my-agent", input=[...])
        return result
        
    except AgentNotFoundError:
        # Agent 不存在，尝试重新注册
        logger.error("Agent not found, attempting to redeploy")
        await redeploy_agent("my-agent")
        
    except TimeoutError:
        # 超时，切换到异步模式
        logger.warning("Sync timeout, switching to async mode")
        task = await client.run_async(agent="my-agent", input=[...])
        return await poll_for_result(task.id)
        
    except AgentError as e:
        # Agent 执行错误，记录并告警
        logger.error(f"Agent execution failed: {e}")
        await send_alert(f"Agent my-agent failed: {e}")
        raise

6. 集成成本分析

6.1 开发成本对比

xychart-beta
    title "不同集成方式的成本对比"
    x-axis ["自定义 API", "SDK 集成", "零代码 HTTP"]
    y-axis "开发时间 (天)"
    bar [14, 3, 1]

6.2 运维成本估算

6.3 ROI 分析

假设一个中型企业有 10 个 Agent 需要集成：

flowchart LR
    subgraph "投资回报分析 (3年)"
        Investment[投资: $80K<br/>SDK + 集成开发]
        Savings[节省: $450K<br/>减少重复开发]
        Net[净收益: $370K<br/>ROI: 462%]
        
        Investment --> Net
        Savings --> Net
    end

7. 迁移到 A2A 的建议

由于 ACP 已合并入 A2A，新项目应考虑直接使用 A2A SDK[4]：

7.1 A2A SDK 生态

7.2 迁移指南

flowchart TB
    subgraph "ACP 到 A2A 迁移路径"
        Step1[1. 评估现有代码] --> Step2[2. 学习 A2A 差异]
        Step2 --> Step3[3. 更新依赖]
        Step3 --> Step4[4. 修改消息格式]
        Step4 --> Step5[5. 测试验证]
        Step5 --> Step6[6. 部署上线]
    end

主要差异：

参考来源

1. ACP GitHub 仓库 - SDK 列表
2. ACP Python SDK 文档
3. ACP 官方文档 - REST API 指南
4. ACP to A2A Migration Guide
5. BeeAI Platform 文档

本文档详细介绍了 ACP 的 SDK 和实现方案。由于 ACP 已归档，建议新项目使用 A2A SDK，现有 ACP 项目可参考迁移指南逐步迁移。