agent_client_protocol/
acp_agent.rs

1//! Utilities for connecting to ACP agents and proxies.
2//!
3//! This module provides [`AcpAgent`], a convenient wrapper around [`crate::schema::v1::McpServer`]
4//! that can be parsed from either a command string or JSON configuration.
5
6use std::path::PathBuf;
7use std::str::FromStr;
8use std::sync::Arc;
9
10use async_process::Child;
11use std::pin::pin;
12
13use crate::schema::v1::{EnvVariable, McpServer as SchemaMcpServer, McpServerStdio};
14use crate::{Client, Conductor, Role};
15
16/// Direction of a line being sent or received.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub enum LineDirection {
19    /// Line being sent to the agent (stdin)
20    Stdin,
21    /// Line being received from the agent (stdout)
22    Stdout,
23    /// Line being received from the agent (stderr)
24    Stderr,
25}
26
27/// A component representing an external ACP agent running in a separate process.
28///
29/// `AcpAgent` implements the [`ConnectTo`](`crate::ConnectTo`) trait for spawning and communicating with
30/// external agents or proxies via stdio. It handles process spawning, stream setup, and
31/// byte stream serialization automatically. This is the primary way to connect to agents
32/// that run as separate executables.
33///
34/// This is a wrapper around [`crate::schema::v1::McpServer`] that provides convenient parsing
35/// from command-line strings or JSON configurations.
36///
37/// # Use Cases
38///
39/// - **External agents**: Connect to agents written in any language (Python, Node.js, Rust, etc.)
40/// - **Proxy chains**: Spawn intermediate proxies that transform or intercept messages
41/// - **Conductor components**: Use with the conductor to build proxy chains
42/// - **Subprocess isolation**: Run potentially untrusted code in a separate process
43///
44/// # Examples
45///
46/// Parse from a command string:
47/// ```
48/// # use agent_client_protocol::AcpAgent;
49/// # use std::str::FromStr;
50/// let agent = AcpAgent::from_str("python my_agent.py --verbose").unwrap();
51/// ```
52///
53/// Parse from JSON:
54/// ```
55/// # use agent_client_protocol::AcpAgent;
56/// # use std::str::FromStr;
57/// let agent = AcpAgent::from_str(r#"{"type": "stdio", "name": "my-agent", "command": "python", "args": ["my_agent.py"], "env": []}"#).unwrap();
58/// ```
59pub struct AcpAgent {
60    server: SchemaMcpServer,
61    debug_callback: Option<Arc<dyn Fn(&str, LineDirection) + Send + Sync + 'static>>,
62}
63
64impl std::fmt::Debug for AcpAgent {
65    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        f.debug_struct("AcpAgent")
67            .field("server", &self.server)
68            .field(
69                "debug_callback",
70                &self.debug_callback.as_ref().map(|_| "..."),
71            )
72            .finish()
73    }
74}
75
76impl AcpAgent {
77    /// Create a new `AcpAgent` from an [`crate::schema::v1::McpServer`] configuration.
78    #[must_use]
79    pub fn new(server: SchemaMcpServer) -> Self {
80        Self {
81            server,
82            debug_callback: None,
83        }
84    }
85
86    /// Create an ACP agent for the Claude Agent adapter.
87    /// Just runs `npx -y @agentclientprotocol/claude-agent-acp@latest`.
88    #[must_use]
89    pub fn claude_agent() -> Self {
90        Self::from_str("npx -y @agentclientprotocol/claude-agent-acp@latest")
91            .expect("valid bash command")
92    }
93
94    /// Create an ACP agent for the Codex adapter.
95    /// Just runs `npx -y @agentclientprotocol/codex-acp@latest`.
96    #[must_use]
97    pub fn codex() -> Self {
98        Self::from_str("npx -y @agentclientprotocol/codex-acp@latest").expect("valid bash command")
99    }
100
101    /// Create an ACP agent for Zed Industries' Claude Code tool.
102    /// Just runs `npx -y @zed-industries/claude-code-acp@latest`.
103    #[deprecated(
104        since = "1.2.0",
105        note = "the package moved to @agentclientprotocol/claude-agent-acp; use `AcpAgent::claude_agent()` instead"
106    )]
107    #[must_use]
108    pub fn zed_claude_code() -> Self {
109        Self::from_str("npx -y @zed-industries/claude-code-acp@latest").expect("valid bash command")
110    }
111
112    /// Create an ACP agent for Zed Industries' Codex tool.
113    /// Just runs `npx -y @zed-industries/codex-acp@latest`.
114    #[deprecated(
115        since = "1.2.0",
116        note = "the package moved to @agentclientprotocol/codex-acp; use `AcpAgent::codex()` instead"
117    )]
118    #[must_use]
119    pub fn zed_codex() -> Self {
120        Self::from_str("npx -y @zed-industries/codex-acp@latest").expect("valid bash command")
121    }
122
123    /// Create an ACP agent for Google's Gemini CLI.
124    /// Just runs `npx -y -- @google/gemini-cli@latest --experimental-acp`.
125    #[must_use]
126    pub fn google_gemini() -> Self {
127        Self::from_str("npx -y -- @google/gemini-cli@latest --experimental-acp")
128            .expect("valid bash command")
129    }
130
131    /// Get the underlying [`crate::schema::v1::McpServer`] configuration.
132    #[must_use]
133    pub fn server(&self) -> &SchemaMcpServer {
134        &self.server
135    }
136
137    /// Convert into the underlying [`crate::schema::v1::McpServer`] configuration.
138    #[must_use]
139    pub fn into_server(self) -> SchemaMcpServer {
140        self.server
141    }
142
143    /// Add a debug callback that will be invoked for each line sent/received.
144    ///
145    /// The callback receives the line content and the direction (stdin/stdout/stderr).
146    /// This is useful for logging, debugging, or monitoring agent communication.
147    ///
148    /// # Example
149    ///
150    /// ```no_run
151    /// # use agent_client_protocol::{AcpAgent, LineDirection};
152    /// # use std::str::FromStr;
153    /// let agent = AcpAgent::from_str("python my_agent.py")
154    ///     .unwrap()
155    ///     .with_debug(|line, direction| {
156    ///         eprintln!("{:?}: {}", direction, line);
157    ///     });
158    /// ```
159    #[must_use]
160    pub fn with_debug<F>(mut self, callback: F) -> Self
161    where
162        F: Fn(&str, LineDirection) + Send + Sync + 'static,
163    {
164        self.debug_callback = Some(Arc::new(callback));
165        self
166    }
167
168    /// Spawn the process and get stdio streams.
169    /// Used internally by the `ConnectTo` trait implementation.
170    pub fn spawn_process(
171        &self,
172    ) -> Result<
173        (
174            async_process::ChildStdin,
175            async_process::ChildStdout,
176            async_process::ChildStderr,
177            Child,
178        ),
179        crate::Error,
180    > {
181        match &self.server {
182            SchemaMcpServer::Stdio(stdio) => {
183                let mut cmd = async_process::Command::new(&stdio.command);
184                cmd.args(&stdio.args);
185                for env_var in &stdio.env {
186                    cmd.env(&env_var.name, &env_var.value);
187                }
188                #[cfg(windows)]
189                {
190                    use async_process::windows::CommandExt as _;
191
192                    cmd.creation_flags(windows_sys::Win32::System::Threading::CREATE_NO_WINDOW);
193                }
194                cmd.stdin(std::process::Stdio::piped())
195                    .stdout(std::process::Stdio::piped())
196                    .stderr(std::process::Stdio::piped());
197
198                let mut child = cmd.spawn().map_err(crate::Error::into_internal_error)?;
199
200                let child_stdin = child
201                    .stdin
202                    .take()
203                    .ok_or_else(|| crate::util::internal_error("Failed to open stdin"))?;
204                let child_stdout = child
205                    .stdout
206                    .take()
207                    .ok_or_else(|| crate::util::internal_error("Failed to open stdout"))?;
208                let child_stderr = child
209                    .stderr
210                    .take()
211                    .ok_or_else(|| crate::util::internal_error("Failed to open stderr"))?;
212
213                Ok((child_stdin, child_stdout, child_stderr, child))
214            }
215            SchemaMcpServer::Http(_) => Err(crate::util::internal_error(
216                "HTTP transport not yet supported by AcpAgent",
217            )),
218            SchemaMcpServer::Sse(_) => Err(crate::util::internal_error(
219                "SSE transport not yet supported by AcpAgent",
220            )),
221            _ => Err(crate::util::internal_error(
222                "Unknown MCP server transport type",
223            )),
224        }
225    }
226}
227
228/// A wrapper around Child that kills the process when dropped.
229struct ChildGuard(Child);
230
231impl ChildGuard {
232    async fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
233        self.0.status().await
234    }
235}
236
237impl Drop for ChildGuard {
238    fn drop(&mut self) {
239        drop(self.0.kill());
240    }
241}
242
243/// Waits for a child process and returns an error if it exits with non-zero status.
244///
245/// The error message includes any stderr output collected concurrently.
246/// When dropped, the child process is killed.
247async fn monitor_child(
248    child: Child,
249    stderr_rx: futures::channel::oneshot::Receiver<String>,
250) -> Result<(), crate::Error> {
251    let mut guard = ChildGuard(child);
252
253    let status = guard
254        .wait()
255        .await
256        .map_err(|e| crate::util::internal_error(format!("Failed to wait for process: {e}")))?;
257
258    if status.success() {
259        Ok(())
260    } else {
261        let stderr = stderr_rx.await.unwrap_or_default();
262
263        let message = if stderr.is_empty() {
264            format!("Process exited with {status}")
265        } else {
266            format!("Process exited with {status}: {stderr}")
267        };
268
269        Err(crate::util::internal_error(message))
270    }
271}
272
273/// Roles that an ACP agent executable can potentially serve.
274pub trait AcpAgentCounterpartRole: Role {}
275
276impl AcpAgentCounterpartRole for Client {}
277
278impl AcpAgentCounterpartRole for Conductor {}
279
280impl<Counterpart: AcpAgentCounterpartRole> crate::ConnectTo<Counterpart> for AcpAgent {
281    async fn connect_to(
282        self,
283        client: impl crate::ConnectTo<Counterpart::Counterpart>,
284    ) -> Result<(), crate::Error> {
285        use futures::io::BufReader;
286        use futures::{AsyncBufReadExt, AsyncWriteExt, StreamExt};
287
288        let (child_stdin, child_stdout, child_stderr, child) = self.spawn_process()?;
289
290        // Create a channel to collect stderr for error reporting
291        let (stderr_tx, stderr_rx) = futures::channel::oneshot::channel::<String>();
292
293        // Read stderr concurrently, optionally calling the debug callback.
294        // We use futures::future::select below to race this against the protocol,
295        // so this runs as part of the same task — no tokio::spawn needed.
296        let debug_callback = self.debug_callback.clone();
297        let stderr_future = async move {
298            let stderr_reader = BufReader::new(child_stderr);
299            let mut stderr_lines = stderr_reader.lines();
300            let mut collected = String::new();
301            while let Some(line_result) = stderr_lines.next().await {
302                if let Ok(line) = line_result {
303                    if let Some(ref callback) = debug_callback {
304                        callback(&line, LineDirection::Stderr);
305                    }
306                    if !collected.is_empty() {
307                        collected.push('\n');
308                    }
309                    collected.push_str(&line);
310                }
311            }
312            drop(stderr_tx.send(collected));
313        };
314
315        // Create a future that monitors the child process for early exit
316        let child_monitor = monitor_child(child, stderr_rx);
317
318        // Convert stdio to line streams with optional debug inspection
319        let incoming_lines: std::pin::Pin<
320            Box<dyn futures::Stream<Item = std::io::Result<String>> + Send>,
321        > = if let Some(callback) = self.debug_callback.clone() {
322            Box::pin(BufReader::new(child_stdout).lines().inspect(move |result| {
323                if let Ok(line) = result {
324                    callback(line, LineDirection::Stdout);
325                }
326            }))
327        } else {
328            Box::pin(BufReader::new(child_stdout).lines())
329        };
330
331        // Create a sink that writes lines (with newlines) to stdin with optional debug logging
332        let outgoing_sink: std::pin::Pin<
333            Box<dyn futures::Sink<String, Error = std::io::Error> + Send>,
334        > = if let Some(callback) = self.debug_callback.clone() {
335            Box::pin(futures::sink::unfold(
336                (child_stdin, callback),
337                async move |(mut writer, callback), line: String| {
338                    callback(&line, LineDirection::Stdin);
339                    let mut bytes = line.into_bytes();
340                    bytes.push(b'\n');
341                    writer.write_all(&bytes).await?;
342                    Ok::<_, std::io::Error>((writer, callback))
343                },
344            ))
345        } else {
346            Box::pin(futures::sink::unfold(
347                child_stdin,
348                async move |mut writer, line: String| {
349                    let mut bytes = line.into_bytes();
350                    bytes.push(b'\n');
351                    writer.write_all(&bytes).await?;
352                    Ok::<_, std::io::Error>(writer)
353                },
354            ))
355        };
356
357        // Race the protocol against child process exit.
358        // Also run stderr collection concurrently.
359        let protocol_future = crate::ConnectTo::<Counterpart>::connect_to(
360            crate::Lines::new(outgoing_sink, incoming_lines),
361            client,
362        );
363
364        let stderr_future = pin!(stderr_future);
365        let protocol_future = pin!(protocol_future);
366        let child_monitor = pin!(child_monitor);
367
368        // Run stderr reader alongside the main race
369        let main_race = async {
370            match futures::future::select(protocol_future, child_monitor).await {
371                futures::future::Either::Left((result, _))
372                | futures::future::Either::Right((result, _)) => result,
373            }
374        };
375
376        // Run stderr collection concurrently with the main logic.
377        // When main_race completes, we don't need stderr anymore.
378        let main_race = pin!(main_race);
379        match futures::future::select(main_race, stderr_future).await {
380            futures::future::Either::Left((result, _)) => result,
381            futures::future::Either::Right(((), protocol)) => protocol.await,
382        }
383    }
384}
385
386impl AcpAgent {
387    /// Create an `AcpAgent` from an iterator of command-line arguments.
388    ///
389    /// Leading arguments of the form `NAME=value` are parsed as environment variables.
390    /// The first non-env argument is the command, and the rest are arguments.
391    ///
392    /// # Example
393    ///
394    /// ```
395    /// # use agent_client_protocol::AcpAgent;
396    /// let agent = AcpAgent::from_args([
397    ///     "RUST_LOG=debug",
398    ///     "cargo",
399    ///     "run",
400    ///     "-p",
401    ///     "my-crate",
402    /// ]).unwrap();
403    /// ```
404    pub fn from_args<I, T>(args: I) -> Result<Self, crate::Error>
405    where
406        I: IntoIterator<Item = T>,
407        T: ToString,
408    {
409        let args: Vec<String> = args.into_iter().map(|s| s.to_string()).collect();
410
411        if args.is_empty() {
412            return Err(crate::util::internal_error("Arguments cannot be empty"));
413        }
414
415        let mut env = vec![];
416        let mut command_idx = 0;
417
418        for (i, arg) in args.iter().enumerate() {
419            if let Some((name, value)) = parse_env_var(arg) {
420                env.push(EnvVariable::new(name, value));
421                command_idx = i + 1;
422            } else {
423                break;
424            }
425        }
426
427        if command_idx >= args.len() {
428            return Err(crate::util::internal_error(
429                "No command found (only environment variables provided)",
430            ));
431        }
432
433        let command = PathBuf::from(&args[command_idx]);
434        let cmd_args = args[command_idx + 1..].to_vec();
435
436        let name = command
437            .file_name()
438            .and_then(|n| n.to_str())
439            .unwrap_or("agent")
440            .to_string();
441
442        Ok(AcpAgent {
443            server: SchemaMcpServer::Stdio(
444                McpServerStdio::new(name, command).args(cmd_args).env(env),
445            ),
446            debug_callback: None,
447        })
448    }
449}
450
451/// Parse a string as an environment variable assignment (NAME=value).
452fn parse_env_var(s: &str) -> Option<(String, String)> {
453    let eq_pos = s.find('=')?;
454    if eq_pos == 0 {
455        return None;
456    }
457
458    let name = &s[..eq_pos];
459    let value = &s[eq_pos + 1..];
460
461    let mut chars = name.chars();
462    let first = chars.next()?;
463    if !first.is_ascii_alphabetic() && first != '_' {
464        return None;
465    }
466    if !chars.all(|c| c.is_ascii_alphanumeric() || c == '_') {
467        return None;
468    }
469
470    Some((name.to_string(), value.to_string()))
471}
472
473impl FromStr for AcpAgent {
474    type Err = crate::Error;
475
476    fn from_str(s: &str) -> Result<Self, Self::Err> {
477        let trimmed = s.trim();
478
479        if trimmed.starts_with('{') {
480            let server: SchemaMcpServer = serde_json::from_str(trimmed)
481                .map_err(|e| crate::util::internal_error(format!("Failed to parse JSON: {e}")))?;
482            return Ok(Self {
483                server,
484                debug_callback: None,
485            });
486        }
487
488        let parts = shell_words::split(trimmed)
489            .map_err(|e| crate::util::internal_error(format!("Failed to parse command: {e}")))?;
490
491        Self::from_args(parts)
492    }
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498
499    #[test]
500    fn test_parse_simple_command() {
501        let agent = AcpAgent::from_str("python agent.py").unwrap();
502        match agent.server {
503            SchemaMcpServer::Stdio(stdio) => {
504                assert_eq!(stdio.name, "python");
505                assert_eq!(stdio.command, PathBuf::from("python"));
506                assert_eq!(stdio.args, vec!["agent.py"]);
507                assert!(stdio.env.is_empty());
508            }
509            _ => panic!("Expected Stdio variant"),
510        }
511    }
512
513    #[test]
514    fn test_parse_command_with_args() {
515        let agent = AcpAgent::from_str("node server.js --port 8080 --verbose").unwrap();
516        match agent.server {
517            SchemaMcpServer::Stdio(stdio) => {
518                assert_eq!(stdio.name, "node");
519                assert_eq!(stdio.command, PathBuf::from("node"));
520                assert_eq!(stdio.args, vec!["server.js", "--port", "8080", "--verbose"]);
521                assert!(stdio.env.is_empty());
522            }
523            _ => panic!("Expected Stdio variant"),
524        }
525    }
526
527    #[test]
528    fn test_parse_command_with_quotes() {
529        let agent = AcpAgent::from_str(r#"python "my agent.py" --name "Test Agent""#).unwrap();
530        match agent.server {
531            SchemaMcpServer::Stdio(stdio) => {
532                assert_eq!(stdio.name, "python");
533                assert_eq!(stdio.command, PathBuf::from("python"));
534                assert_eq!(stdio.args, vec!["my agent.py", "--name", "Test Agent"]);
535                assert!(stdio.env.is_empty());
536            }
537            _ => panic!("Expected Stdio variant"),
538        }
539    }
540
541    #[test]
542    fn test_parse_json_stdio() {
543        let json = r#"{
544            "type": "stdio",
545            "name": "my-agent",
546            "command": "/usr/bin/python",
547            "args": ["agent.py", "--verbose"],
548            "env": []
549        }"#;
550        let agent = AcpAgent::from_str(json).unwrap();
551        match agent.server {
552            SchemaMcpServer::Stdio(stdio) => {
553                assert_eq!(stdio.name, "my-agent");
554                assert_eq!(stdio.command, PathBuf::from("/usr/bin/python"));
555                assert_eq!(stdio.args, vec!["agent.py", "--verbose"]);
556                assert!(stdio.env.is_empty());
557            }
558            _ => panic!("Expected Stdio variant"),
559        }
560    }
561
562    #[test]
563    fn test_parse_json_http() {
564        let json = r#"{
565            "type": "http",
566            "name": "remote-agent",
567            "url": "https://example.com/agent",
568            "headers": []
569        }"#;
570        let agent = AcpAgent::from_str(json).unwrap();
571        match agent.server {
572            SchemaMcpServer::Http(http) => {
573                assert_eq!(http.name, "remote-agent");
574                assert_eq!(http.url, "https://example.com/agent");
575                assert!(http.headers.is_empty());
576            }
577            _ => panic!("Expected Http variant"),
578        }
579    }
580}
agent_client_protocol/acp_agent.rs

agent_client_protocol/
acp_agent.rs
