基于 SACP 的 Rust SDK

let mut results = Vec::new();

let _: () = patchwork.think()
    .text("处理每个项目并使用 `record` 工具记录它")
    .tool(
        "record",
        "记录处理过的项目",
        async |input: RecordInput, _cx| {
            results.push(input.item);
            Ok(RecordOutput { success: true })
        },
        acp::tool_fn_mut!(),
    )
    .await?;

// 在 think 块之后，`results` 包含所有记录的项目
println!("Recorded: {:?}", results);

#[async_trait]
pub trait Agent {
    async fn prompt(&self, args: PromptRequest) -> Result<PromptResponse>;
    // ...
}

// 有问题的：消息处理在处理时饥饿
while let Some(message) = connection.next().await {
    process(message).await; // 在此等待期间连接是静止的！
}

// 正确：消息处理并发继续
connection.run_until(async |cx| {
    // 此代码运行时连接处理消息
    let response = cx.send_request(request).block_task().await?;
    process(response).await
}).await

// 作为客户端连接到代理
ClientToAgent::builder()
    .name("my-client") // 可选，对追踪有用

// 作为代理服务客户端
AgentToClient::builder()
    .name("my-agent") // 可选，对追踪有用

// 通过 stdio 连接到代理
ClientToAgent::builder()
    .connect_to(acp::stdio())

// 通过命令连接到特定代理
ClientToAgent::builder()
    .connect_to(AcpAgent::from_str("some-command --acp"))

// 连接到 Zed 的 Claude Code 集成
ClientToAgent::builder()
    .connect_to(AcpAgent::zed_claude_code())

ClientToAgent::builder()
    .connect_to(AgentToClient::builder())

impl Component<AgentToClient> for MyAgent {
    async fn serve(self, client: impl Component<ClientToAgent>) -> Result<(), acp::Error> {
        AgentToClient::builder()
            .on_receive_request(/* ... */)
            .serve(client)
            .await
    }
}

// 客户端直接连接到代理——对测试有用
ClientToAgent::builder()
    .connect_to(MyAgent::new())

AgentToClient::builder()
    .name("my-agent")
    .on_receive_request(/* ... */)
    .connect_to(transport)?
    .serve()
    .await

ClientToAgent::builder()
    .name("my-client")
    .connect_to(transport)?
    .run_until(async |cx| {
        // 你的代码在这里运行，而消息在后台处理。
        // 使用 `cx` 发送请求和通知。
        let response = cx.send_request(InitializeRequest::new(ProtocolVersion::LATEST))
            .block_task().await?;
        Ok(response)
    })
    .await

cx.send_notification(StatusNotification::new("processing"))?;

let response_handle = cx.send_request(PromptRequest::new(session_id, messages));

cx.send_request(PromptRequest::new(session_id, messages))
    .on_ok_response(
        async move |response: PromptResponse, cx| {
            println!("代理完成: {:?}", response.stop_reason);
            Ok(())
        },
        acp::on_response!()
    )?;

let response: PromptResponse = cx.send_request(PromptRequest::new(session_id, messages))
    .block_task()
    .await?;

.on_receive_request(async |req: PromptRequest, request_cx, cx| {
    // 在此运行期间不会处理其他消息
    cx.send_notification(StatusNotification::new("processing"))?;
    // 通知保证在响应之前发送
    request_cx.respond(PromptResponse::new(StopReason::EndTurn))
}, acp::on_receive_request!())

// 错误——会死锁
.on_receive_request(async |req: PromptRequest, request_cx, cx| {
    let response = cx.send_request(SomeOtherRequest::new())
        .block_task()  // 死锁！处理程序阻塞等待响应
        .await?;       // 但响应在被处理程序返回之前无法被处理
    request_cx.respond(/* ... */)
}, acp::on_receive_request!())

.on_receive_request(async |request: PromptRequest, request_cx, cx| {
    cx.spawn(async move {
        // 在这里使用 block_task 是安全的——我们在生成的任务中
        let response = cx.send_request(InitializeRequest::new(ProtocolVersion::LATEST))
            .block_task()
            .await?;
        /* ... */
        Ok(())
    })?;

    // 处理程序立即返回，生成的工作继续
    request_cx.respond(/* ... */)
}, acp::on_receive_request!())

cx.build_session("/path/to/workspace")
    .with_mcp_server(my_mcp_server)?    // 附加 MCP 服务器（见下文）
    // ... 其他配置

// 在 run_until 闭包内部（不是处理程序）
let workspace_path = Path::new("/my/workspace");

cx.build_session(workspace_path)
    .with_mcp_server(
        McpServer::builder("tools")
            .tool_fn("get_path", "返回路径", async move |_: (), _cx| {
                // 可以按引用捕获 `workspace_path`！
                Ok(workspace_path.display().to_string())
            }, acp::tool_fn!())
            .build()
    )?
    .block_task()
    .run_until(async |mut session| {
        session.send_prompt("工作区路径是什么？")?;
        let response = session.read_to_string().await?;
        println!("{response}");
        Ok(())
    })
    .await?;

use agent_client_protocol_rmcp::RmcpServer;

cx.build_session(workspace_path)
    .with_mcp_server(RmcpServer::new(my_rmcp_service))?
    // ...

.on_receive_request(async |req: NewSessionRequest, request_cx, cx| {
    cx.build_session_from(req)
        .with_mcp_server(my_mcp_server)?
        .on_session_start(async |mut session| {
            session.send_prompt("Hello")?;
            let response = session.read_to_string().await?;
            Ok(())
        })?;

    // 处理程序立即返回，会话在后台运行
    Ok(())
}, acp::on_receive_request!())

let mut session = cx.build_session(workspace_path)
    .with_mcp_server(my_mcp_server)?
    .block_task()
    .start_session()
    .await?;

session.send_prompt("Hello")?;
let response = session.read_to_string().await?;

.on_receive_request(async |req: NewSessionRequest, request_cx, cx| {
    let session_id = cx.build_session_from(req)
        .with_mcp_server(injected_tools)?
        .block_task()
        .start_session_proxy(request_cx)
        .await?;

    // 会话消息在客户端和代理之间自动代理
    Ok(())
}, acp::on_receive_request!())

AgentToClient::builder()
    .on_receive_notification(
        async |notif: SessionNotification, cx| {
            //        -------------------
            //        预期的通知类型
            println!("会话更新: {:?}", notif.update);
            Ok(())
        },
        acp::on_receive_notification!(), // <-- 需要技巧性的宏参数
    )

.on_receive_request(
    async |req: PromptRequest, request_cx, cx| {
        // 处理请求...
        cx.send_notification(StatusNotification::new("processing"))?;

        // 发送响应
        request_cx.respond(PromptResponse::new(StopReason::EndTurn))
    },
    acp::on_receive_request!(),
)

.on_receive_request(
    async |req: LongRunningRequest, request_cx, cx| {
        cx.spawn(async move {
            let result = do_expensive_work(&req).await;
            request_cx.respond_with_result(result)
        });
        Ok(())
    },
    acp::on_receive_request!(),
)

#[derive(Debug, Serialize, Deserialize, JsonSchema, JrNotification)]
#[notification(method = "_myapp/progress")]
struct ProgressNotification {
    percent: u32,
    message: String,
}

#[derive(Debug, Serialize, Deserialize, JsonSchema, JrRequest)]
#[request(method = "_myapp/compute", response = ComputeResponse)]
struct ComputeRequest {
    input: String,
}

#[derive(Debug, Serialize, Deserialize, JsonSchema)]
struct ComputeResponse {
    result: String,
}

ClientToAgent::builder()
    .on_receive_notification(
        async |notif: ProgressNotification, cx| {
            println!("进度: {}% - {}", notif.percent, notif.message);
            Ok(())
        },
        acp::on_receive_notification!()
    )

.on_receive_message(
    async |message: MessageCx, cx| {
        // 将所有消息转发到另一个线程进行处理
        message_sender.send(message)?;
        Ok(())
    },
    acp::on_receive_message!(),
)

pub trait JrMessageHandler: Send {
    type Link: JrLink;

    async fn handle_message(
        &mut self,
        message: MessageCx,
        cx: JrConnectionCx<Self::Link>,
    ) -> Result<Handled<MessageCx>, Error>;
}

pub enum Handled<T> {
    Yes,
    No { message: T, retry: bool },
}

.on_receive_request(async |mut req: EchoRequest, request_cx, cx| {
    req.text.push("modified".to_string());
    Ok(Handled::No {
        message: (req, request_cx),
        retry: false,
    })
}, acp::on_receive_request!())

let registration = cx.add_dynamic_handler(MySessionHandler::new(session_id))?;

registration.run_indefinitely();

impl JrMessageHandler for MyHandler {
    type Link = AgentToClient;

    async fn handle_message(
        &mut self,
        message: MessageCx,
        cx: JrConnectionCx<Self::Link>,
    ) -> Result<Handled<MessageCx>, Error> {
        MatchMessage::new(message)
            .if_request(async |req: InitializeRequest, request_cx| {
                request_cx.respond(InitializeResponse::new(req.protocol_version))
            })
            .if_request(async |req: PromptRequest, request_cx| {
                request_cx.respond(PromptResponse::new(StopReason::EndTurn))
            })
            .if_notification(async |notif: SessionNotification| {
                log::info!("会话更新: {:?}", notif);
                Ok(())
            })
            .await
            .done()
    }
}