acp_utils/client/
tokio_agent.rs

1//! Tokio-native parent-side ACP transport.
2//!
3//! `agent_client_protocol::AcpAgent` spawns the child via smol's
4//! `async_process::Command`, which wraps stdio in `blocking::Unblock`. Inside a
5//! tokio runtime that causes a busy loop. This avoids the issue by spawning stdio agents with `tokio::process::Command`
6//!
7use agent_client_protocol::schema::{McpServer, McpServerStdio};
8use agent_client_protocol::util::internal_error;
9use agent_client_protocol::{AcpAgent, ByteStreams, ConnectTo, Error, Role, util};
10use std::path::PathBuf;
11use std::process::Stdio;
12use std::str::FromStr;
13use tokio::io::{AsyncBufReadExt, BufReader};
14use tokio::process::Command;
15use tokio::sync::oneshot;
16use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
17
18pub struct TokioAcpAgent {
19    stdio: McpServerStdio,
20}
21
22impl TokioAcpAgent {
23    pub fn from_command(command: impl Into<PathBuf>, args: Vec<String>) -> Self {
24        let command = command.into();
25        let name = command.file_name().and_then(|name| name.to_str()).unwrap_or("acp-agent").to_string();
26        let mut stdio = McpServerStdio::new(name, command);
27        stdio.args = args;
28        Self { stdio }
29    }
30
31    pub fn stdio(&self) -> &McpServerStdio {
32        &self.stdio
33    }
34}
35
36impl<T: Role> ConnectTo<T> for TokioAcpAgent {
37    async fn connect_to(self, client: impl ConnectTo<T::Counterpart>) -> Result<(), Error> {
38        connect_stdio::<T>(self.stdio, client).await
39    }
40}
41
42impl FromStr for TokioAcpAgent {
43    type Err = Error;
44
45    fn from_str(s: &str) -> Result<Self, Self::Err> {
46        match AcpAgent::from_str(s)?.into_server() {
47            McpServer::Stdio(stdio) => Ok(Self { stdio }),
48            _ => Err(util::internal_error("unsupported ACP agent transport")),
49        }
50    }
51}
52
53async fn connect_stdio<T: Role>(server: McpServerStdio, client: impl ConnectTo<T::Counterpart>) -> Result<(), Error> {
54    let (stdin, stdout, stderr, mut child) = {
55        let mut cmd = Command::new(&server.command);
56        cmd.args(&server.args);
57        for env_var in &server.env {
58            cmd.env(&env_var.name, &env_var.value);
59        }
60
61        let mut child = cmd
62            .stdin(Stdio::piped())
63            .stdout(Stdio::piped())
64            .stderr(Stdio::piped())
65            .kill_on_drop(true)
66            .spawn()
67            .map_err(Error::into_internal_error)?;
68
69        let stdin = child.stdin.take().ok_or_else(|| internal_error("missing child stdin"))?;
70        let stdout = child.stdout.take().ok_or_else(|| internal_error("missing child stdout"))?;
71        let stderr = child.stderr.take().ok_or_else(|| internal_error("missing child stderr"))?;
72        (stdin, stdout, stderr, child)
73    };
74
75    let (stderr_tx, stderr_rx) = oneshot::channel::<String>();
76    tokio::spawn(async move {
77        let mut lines = BufReader::new(stderr).lines();
78        let mut buf = String::new();
79        while let Ok(Some(line)) = lines.next_line().await {
80            if !buf.is_empty() {
81                buf.push('\n');
82            }
83            buf.push_str(&line);
84        }
85        let _ = stderr_tx.send(buf);
86    });
87
88    let child_fut = async move {
89        match child.wait().await {
90            Ok(s) if s.success() => Ok(()),
91            Ok(s) => {
92                let stderr = stderr_rx.await.unwrap_or_default();
93                Err(util::internal_error(format!("agent process exited ({s}): {stderr}")))
94            }
95            Err(e) => Err(Error::into_internal_error(e)),
96        }
97    };
98
99    let bytes = ByteStreams::new(stdin.compat_write(), stdout.compat());
100    tokio::select! {
101        result = ConnectTo::<T>::connect_to(bytes, client) => result,
102        result = child_fut => result,
103    }
104}
acp_utils/client/tokio_agent.rs

acp_utils/client/
tokio_agent.rs
