agent_call/
lib.rs

1//! One-shot Rust adapters for Claude, Codex, Devin, and OpenCode v2 CLIs.
2//!
3//! Each call runs a fresh CLI process and returns the completed assistant text.
4//! Authentication and model defaults come from the installed CLI unless the
5//! caller chooses a model explicitly.
6
7use std::ffi::OsString;
8use std::path::{Path, PathBuf};
9use std::process::Stdio;
10use std::time::{Duration, Instant};
11
12use serde_json::Value;
13use thiserror::Error;
14use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader};
15use tokio::process::Command;
16
17mod other;
18pub use other::{AgentCli, Provider, ProviderError};
19
20pub(crate) use other::validate_request;
21
22const STDERR_LIMIT: usize = 4096;
23
24/// Which tools the agent may use during a run.
25///
26/// [`ToolPolicy::Default`] keeps each CLI's historical settings (see the README). The other
27/// policies never give the agent a shell, file access or code execution: a CLI that cannot
28/// enforce a requested policy fails with [`Error::InvalidRequest`] instead of running with more
29/// power.
30#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
31#[non_exhaustive]
32pub enum ToolPolicy {
33    /// The CLI's previous behaviour (Claude: `bypassPermissions`; Codex: read-only sandbox).
34    #[default]
35    Default,
36    /// Only tools from the request's MCP servers. Supported by Claude and Codex.
37    McpOnly,
38    /// MCP tools plus the CLI's built-in web search and fetch. Supported by Claude and Codex.
39    Web,
40}
41
42/// An MCP server (streamable HTTP) the agent may use during a run.
43#[derive(Debug, Clone, PartialEq, Eq)]
44pub struct McpServer {
45    /// Server name, used in tool names (`mcp__<name>__…`); ASCII letters, digits, `_` and `-`.
46    pub name: String,
47    /// Streamable HTTP endpoint.
48    pub url: String,
49    /// Environment variable holding a bearer token. Put the token itself in [`RunRequest::env`]
50    /// so it never appears in a command line or a file.
51    pub bearer_token_env: Option<String>,
52}
53
54/// Progress reported while a run is going (see [`RunRequest::on_event`]).
55#[derive(Debug, Clone, PartialEq, Eq)]
56#[non_exhaustive]
57pub enum AgentEvent {
58    /// A piece of the answer as it is generated (Claude).
59    TextDelta(String),
60    /// A complete text block or message (Codex, OpenCode).
61    Text(String),
62    /// The agent started using a tool (`WebSearch`, `mcp__kb__kb_grep`, `web_search`…).
63    ToolUse {
64        /// Tool name as the CLI reports it.
65        name: String,
66    },
67}
68
69/// Receives [`AgentEvent`]s; called from the task reading the CLI's output, so keep it quick.
70#[derive(Clone)]
71pub struct EventHandler(pub std::sync::Arc<dyn Fn(AgentEvent) + Send + Sync>);
72
73impl EventHandler {
74    /// Wraps a callback.
75    pub fn new(f: impl Fn(AgentEvent) + Send + Sync + 'static) -> Self {
76        Self(std::sync::Arc::new(f))
77    }
78
79    pub(crate) fn emit(&self, event: AgentEvent) {
80        (self.0)(event);
81    }
82}
83
84impl std::fmt::Debug for EventHandler {
85    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86        f.write_str("EventHandler")
87    }
88}
89
90/// A single, fresh CLI run.
91///
92/// Build it with struct update syntax so new optional fields do not break callers:
93/// `RunRequest { prompt, cwd, ..RunRequest::default() }`.
94#[derive(Debug, Clone)]
95pub struct RunRequest {
96    pub prompt: String,
97    pub cwd: PathBuf,
98    /// Model ID for the chosen provider, or `None` for its configured default.
99    pub model: Option<String>,
100    pub timeout: Duration,
101    /// Environment variable names to remove from the child only.
102    pub env_remove: Vec<OsString>,
103    /// Optional JSON Schema. Claude and Codex enforce it; Devin and OpenCode
104    /// leave validation to the caller.
105    pub json_schema: Option<Value>,
106    /// Which tools the agent may use.
107    pub tool_policy: ToolPolicy,
108    /// MCP servers for this run. Claude and Codex only; with a policy other than
109    /// [`ToolPolicy::Default`], MCP servers from the user's own CLI configuration are ignored.
110    pub mcp_servers: Vec<McpServer>,
111    /// Extra environment variables for the child only, set after `env_remove` is applied.
112    pub env: Vec<(OsString, OsString)>,
113    /// Instructions appended to the CLI's own system prompt (Claude: `--append-system-prompt`,
114    /// Codex: `developer_instructions`). Devin and OpenCode receive it before the prompt.
115    pub system_prompt: Option<String>,
116    /// Keep the CLI's session so a later run can resume it (default: ephemeral, as before 0.4).
117    pub persist_session: bool,
118    /// Continue the session returned earlier in [`RunResult::session_id`]. Implies persistence.
119    /// Claude and Codex resume in the same `cwd`; OpenCode by id; Devin rejects it.
120    pub resume: Option<String>,
121    /// Called with progress while the run goes on (Claude streams text; Codex and OpenCode report
122    /// whole messages and tool calls; Devin reports nothing).
123    pub on_event: Option<EventHandler>,
124}
125
126impl Default for RunRequest {
127    /// An empty prompt in the current directory with a five-minute timeout.
128    fn default() -> Self {
129        Self {
130            prompt: String::new(),
131            cwd: PathBuf::from("."),
132            model: None,
133            timeout: Duration::from_secs(300),
134            env_remove: Vec::new(),
135            json_schema: None,
136            tool_policy: ToolPolicy::Default,
137            mcp_servers: Vec::new(),
138            env: Vec::new(),
139            system_prompt: None,
140            persist_session: false,
141            resume: None,
142            on_event: None,
143        }
144    }
145}
146
147impl RunRequest {
148    /// The prompt with the system prompt in front, for CLIs without a system prompt option.
149    pub(crate) fn prompt_with_system(&self) -> String {
150        match self.system_prompt.as_deref().map(str::trim) {
151            Some(system) if !system.is_empty() => {
152                format!(
153                    "<instructions>\n{system}\n</instructions>\n\n{}",
154                    self.prompt
155                )
156            }
157            _ => self.prompt.clone(),
158        }
159    }
160
161    /// Rejects tool policies and MCP servers for CLIs that cannot enforce them.
162    pub(crate) fn require_default_tools(&self, cli: &str) -> Result<(), Error> {
163        if self.tool_policy != ToolPolicy::Default {
164            return Err(Error::InvalidRequest(format!(
165                "{cli} cannot enforce tool policy {:?}",
166                self.tool_policy
167            )));
168        }
169        if !self.mcp_servers.is_empty() {
170            return Err(Error::InvalidRequest(format!(
171                "{cli} does not support per-run MCP servers"
172            )));
173        }
174        Ok(())
175    }
176}
177
178/// Token accounting reported by the CLI.
179#[derive(Debug, Clone, Default, PartialEq, Eq)]
180pub struct TokenUsage {
181    pub input: u64,
182    pub output: u64,
183    pub reasoning: u64,
184    pub cache_read: u64,
185    pub cache_write: u64,
186}
187
188/// Result of one CLI run.
189#[derive(Debug, Clone)]
190pub struct RunResult {
191    pub text: String,
192    pub session_id: Option<String>,
193    /// `None` when the CLI stream did not contain complete usage data.
194    pub usage: Option<TokenUsage>,
195    pub duration: Duration,
196    pub stderr_tail: String,
197}
198
199#[derive(Debug, Error)]
200pub enum Error {
201    #[error("CLI executable not found: {0}")]
202    NotFound(PathBuf),
203    #[error("failed to probe CLI version: {0}")]
204    VersionProbe(String),
205    #[error("OpenCode v2 required, found: {0}")]
206    UnsupportedVersion(String),
207    #[error("invalid run request: {0}")]
208    InvalidRequest(String),
209    #[error("failed to run CLI: {0}")]
210    Io(#[from] std::io::Error),
211    #[error("CLI timed out after {0:?}")]
212    Timeout(Duration),
213    #[error("CLI exited with {status:?}: {stderr_tail}")]
214    Exit {
215        status: Option<i32>,
216        stderr_tail: String,
217    },
218    #[error("agent reported an error: {0}")]
219    Agent(String),
220    #[error("invalid CLI output: {0}")]
221    Protocol(String),
222    #[error("agent returned no assistant text")]
223    EmptyResponse,
224    /// The session to resume does not exist (expired, other machine, other `cwd`).
225    #[error("session not found: {0}")]
226    SessionNotFound(String),
227}
228
229/// A verified OpenCode v2 executable.
230#[derive(Debug, Clone)]
231pub struct OpenCodeV2 {
232    binary: PathBuf,
233}
234
235impl OpenCodeV2 {
236    /// Find `opencode` on `PATH` and verify that its major version is 2.
237    pub fn discover() -> Result<Self, Error> {
238        Self::with_binary(PathBuf::from("opencode"))
239    }
240
241    /// Use an explicit executable path, useful for isolated deployments and tests.
242    pub fn with_binary(binary: impl Into<PathBuf>) -> Result<Self, Error> {
243        let binary = binary.into();
244        let output = std::process::Command::new(&binary)
245            .arg("--version")
246            .output()
247            .map_err(|e| {
248                if e.kind() == std::io::ErrorKind::NotFound {
249                    Error::NotFound(binary.clone())
250                } else {
251                    Error::VersionProbe(e.to_string())
252                }
253            })?;
254        if !output.status.success() {
255            return Err(Error::VersionProbe(
256                String::from_utf8_lossy(&output.stderr).trim().to_string(),
257            ));
258        }
259        let version = String::from_utf8_lossy(&output.stdout).trim().to_string();
260        let first = version.trim_start_matches('v');
261        if !first.starts_with("2.") {
262            return Err(Error::UnsupportedVersion(version));
263        }
264        Ok(Self { binary })
265    }
266
267    pub fn binary(&self) -> &Path {
268        &self.binary
269    }
270
271    /// Run a fresh prompt. Dropping the future terminates its child process.
272    pub async fn run(&self, request: RunRequest) -> Result<RunResult, Error> {
273        validate_request(&request)?;
274        request.require_default_tools("OpenCode")?;
275        let started = Instant::now();
276        let mut command = Command::new(&self.binary);
277        command.args(["run", "--standalone", "--format", "json"]);
278        if let Some(session) = request.resume.as_deref() {
279            command.arg("--session").arg(session);
280        }
281        if let Some(model) = request.model.as_deref() {
282            if model.trim().is_empty() || !model.contains('/') {
283                return Err(Error::InvalidRequest(format!(
284                    "model must be provider/model[#variant]: {model}"
285                )));
286            }
287            command.arg("--model").arg(model);
288        }
289        for name in &request.env_remove {
290            command.env_remove(name);
291        }
292        command.envs(request.env.iter().map(|(k, v)| (k, v)));
293        command
294            .current_dir(&request.cwd)
295            .stdin(Stdio::piped())
296            .stdout(Stdio::piped())
297            .stderr(Stdio::piped())
298            .kill_on_drop(true);
299
300        let mut child = command.spawn()?;
301        let mut stdin = child.stdin.take().expect("piped stdin");
302        let stdout = child.stdout.take().expect("piped stdout");
303        let stderr = child.stderr.take().expect("piped stderr");
304        let work = async {
305            let input = async {
306                stdin
307                    .write_all(request.prompt_with_system().as_bytes())
308                    .await?;
309                stdin.shutdown().await?;
310                drop(stdin);
311                Ok::<(), std::io::Error>(())
312            };
313            let (input, parsed, stderr_tail, status) = tokio::join!(
314                input,
315                parse_stdout(stdout, request.on_event.clone()),
316                read_stderr_tail(stderr),
317                child.wait()
318            );
319            input?;
320            let parsed = parsed?;
321            let stderr_tail = stderr_tail?;
322            let status = status?;
323            if !status.success() {
324                return Err(Error::Exit {
325                    status: status.code(),
326                    stderr_tail,
327                });
328            }
329            if let Some(error) = parsed.error {
330                return Err(Error::Agent(error));
331            }
332            let text = parsed.text_parts.join("\n").trim().to_string();
333            if text.is_empty() {
334                return Err(Error::EmptyResponse);
335            }
336            let usage = parsed.usage();
337            Ok(RunResult {
338                text,
339                session_id: parsed.session_id,
340                usage,
341                duration: started.elapsed(),
342                stderr_tail,
343            })
344        };
345        match tokio::time::timeout(request.timeout, work).await {
346            Ok(result) => result,
347            Err(_) => {
348                let _ = child.start_kill();
349                let _ = child.wait().await;
350                Err(Error::Timeout(request.timeout))
351            }
352        }
353    }
354}
355
356#[derive(Default)]
357struct Parsed {
358    text_parts: Vec<String>,
359    session_id: Option<String>,
360    usage_sum: TokenUsage,
361    starts: usize,
362    finishes: usize,
363    last_reason: Option<String>,
364    error: Option<String>,
365}
366
367impl Parsed {
368    fn usage(&self) -> Option<TokenUsage> {
369        (self.finishes > 0
370            && self.starts == self.finishes
371            && self.last_reason.as_deref() != Some("tool-calls"))
372        .then(|| self.usage_sum.clone())
373    }
374}
375
376async fn parse_stdout(
377    stdout: impl AsyncRead + Unpin,
378    on_event: Option<EventHandler>,
379) -> Result<Parsed, Error> {
380    let mut lines = BufReader::new(stdout).lines();
381    let mut parsed = Parsed::default();
382    while let Some(line) = lines.next_line().await? {
383        if line.trim().is_empty() {
384            continue;
385        }
386        let event: Value = serde_json::from_str(&line)
387            .map_err(|e| Error::Protocol(format!("{e}; line: {}", tail(&line, 200))))?;
388        if let Some(id) = event["sessionID"].as_str() {
389            parsed.session_id = Some(id.to_string());
390        }
391        let part = &event["part"];
392        match event["type"].as_str() {
393            Some("text") => {
394                if let Some(text) = part["text"].as_str()
395                    && !text.trim().is_empty()
396                {
397                    if let Some(h) = &on_event {
398                        h.emit(AgentEvent::Text(text.to_string()));
399                    }
400                    parsed.text_parts.push(text.to_string());
401                }
402            }
403            Some("tool_use") => {
404                if let (Some(h), Some(name)) = (&on_event, part["tool"].as_str()) {
405                    h.emit(AgentEvent::ToolUse {
406                        name: name.to_string(),
407                    });
408                }
409            }
410            Some("step_start") => parsed.starts += 1,
411            Some("step_finish") => {
412                parsed.finishes += 1;
413                parsed.last_reason = part["reason"].as_str().map(str::to_string);
414                let tokens = &part["tokens"];
415                parsed.usage_sum.input += tokens["input"].as_u64().unwrap_or(0);
416                parsed.usage_sum.output += tokens["output"].as_u64().unwrap_or(0);
417                parsed.usage_sum.reasoning += tokens["reasoning"].as_u64().unwrap_or(0);
418                parsed.usage_sum.cache_read += tokens["cache"]["read"].as_u64().unwrap_or(0);
419                parsed.usage_sum.cache_write += tokens["cache"]["write"].as_u64().unwrap_or(0);
420            }
421            Some("error") => {
422                let message = event["error"]["data"]["message"]
423                    .as_str()
424                    .or_else(|| event["error"]["message"].as_str())
425                    .or_else(|| event["error"]["name"].as_str())
426                    .or_else(|| event["error"].as_str())
427                    .unwrap_or("unknown OpenCode error");
428                parsed.error = Some(message.to_string());
429            }
430            _ => {}
431        }
432    }
433    Ok(parsed)
434}
435
436async fn read_stderr_tail(mut stderr: impl AsyncRead + Unpin) -> Result<String, std::io::Error> {
437    let mut tail_bytes = Vec::new();
438    let mut chunk = [0_u8; 4096];
439    loop {
440        let n = stderr.read(&mut chunk).await?;
441        if n == 0 {
442            break;
443        }
444        tail_bytes.extend_from_slice(&chunk[..n]);
445        if tail_bytes.len() > STDERR_LIMIT {
446            tail_bytes.drain(..tail_bytes.len() - STDERR_LIMIT);
447        }
448    }
449    Ok(String::from_utf8_lossy(&tail_bytes).to_string())
450}
451
452fn tail(s: &str, count: usize) -> String {
453    let chars: Vec<char> = s.chars().collect();
454    chars[chars.len().saturating_sub(count)..].iter().collect()
455}
agent_call/lib.rs

agent_call/
lib.rs
