ai_crew_sync/client/
mod.rs

1//! Console client: everything the MCP tools can do, from a human terminal.
2//!
3//! Connects to the bus over the same Streamable HTTP transport and the same
4//! bearer token a coding agent would use, so a human on the shell is
5//! just another agent on the bus:
6//!
7//! ```text
8//! export BUS_URL=https://bus.internal.example/mcp
9//! export BUS_TOKEN=acs_...
10//! ai-crew-sync client whoami
11//! ai-crew-sync client send --channel deploys --body "staging is on 1.4.2"
12//! ai-crew-sync client read --scope inbox
13//! ai-crew-sync client task claim --key refactor-auth
14//! ```
15
16use anyhow::{Context, bail};
17use clap::{Args, Subcommand};
18use rmcp::{
19    ServiceExt,
20    model::{CallToolRequestParams, ClientConfig},
21    transport::{
22        StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig,
23    },
24};
25use serde_json::{Value, json};
26
27#[derive(Args)]
28pub struct ClientArgs {
29    /// URL of the bus MCP endpoint, e.g. https://bus.example.com/mcp
30    #[arg(long, env = "BUS_URL", default_value = "http://localhost:8787/mcp")]
31    pub url: String,
32
33    /// Your agent token (issued with `ai-crew-sync token issue`).
34    #[arg(long, env = "BUS_TOKEN", hide_env_values = true)]
35    pub token: String,
36
37    /// Which working context this is — usually the repository name. Separates
38    /// your presence, task claims and locks from your other sessions. Omit it
39    /// to share one context with them.
40    #[arg(long, env = "BUS_SESSION")]
41    pub session: Option<String>,
42
43    /// Print raw JSON instead of the human-readable rendering.
44    #[arg(long, global = true)]
45    pub json: bool,
46
47    #[command(subcommand)]
48    pub command: ClientCmd,
49}
50
51#[derive(Subcommand)]
52pub enum ClientCmd {
53    /// Who am I on the bus, and is anything waiting for me?
54    Whoami,
55    /// List the tools the server exposes (sanity check).
56    Tools,
57    /// Send a message: --channel to broadcast, --to for a direct message.
58    Send {
59        #[arg(long, conflicts_with = "to")]
60        channel: Option<String>,
61        #[arg(long)]
62        to: Option<String>,
63        #[arg(long)]
64        body: String,
65        /// Interrupt every session in the team, not just those watching this
66        /// channel. For deploys, migrations and breaking changes.
67        #[arg(long)]
68        announce: bool,
69        #[arg(long)]
70        reply_to: Option<i64>,
71        /// Attach a file (repeatable). Max 8 files, 256 KiB each.
72        #[arg(long)]
73        file: Vec<std::path::PathBuf>,
74    },
75    /// Attach a file to a task.
76    Attach {
77        /// Task key.
78        task: String,
79        #[arg(long)]
80        file: std::path::PathBuf,
81        #[arg(long)]
82        content_type: Option<String>,
83    },
84    /// Download an attachment by id.
85    Download {
86        id: i64,
87        /// Output path; defaults to the attachment's filename.
88        #[arg(long)]
89        out: Option<std::path::PathBuf>,
90    },
91    /// Ask a teammate's agent and wait for the answer.
92    Ask {
93        to: String,
94        /// The question. Omit when resuming with --resume-id.
95        question: Option<String>,
96        #[arg(long)]
97        timeout_seconds: Option<i64>,
98        /// Keep waiting on an earlier question (its question_message_id).
99        #[arg(long)]
100        resume_id: Option<i64>,
101    },
102    /// Read messages ("all", "inbox", or a channel name).
103    Read {
104        #[arg(long, default_value = "all")]
105        scope: String,
106        /// Re-read history instead of only unread messages.
107        #[arg(long)]
108        history: bool,
109        #[arg(long, default_value_t = 50)]
110        limit: i64,
111        /// Include direct messages addressed to your other sessions.
112        #[arg(long)]
113        all_sessions: bool,
114    },
115    /// Full-text search messages.
116    Search {
117        query: String,
118        #[arg(long, default_value_t = 50)]
119        limit: i64,
120    },
121    /// List channels.
122    Channels,
123    /// Create a channel.
124    ChannelCreate {
125        name: String,
126        #[arg(long)]
127        topic: Option<String>,
128    },
129    /// Who is on the bus and what are they doing?
130    Agents {
131        #[arg(long)]
132        online: bool,
133    },
134    /// Publish your own presence.
135    Beat {
136        #[arg(long)]
137        status: Option<String>,
138        #[arg(long)]
139        repo: Option<String>,
140        #[arg(long)]
141        branch: Option<String>,
142        #[arg(long)]
143        activity: Option<String>,
144        #[arg(long)]
145        ttl_seconds: Option<i64>,
146    },
147    /// List tasks.
148    Tasks {
149        #[arg(long)]
150        status: Option<String>,
151        #[arg(long)]
152        mine: bool,
153    },
154    /// Operate on a single task.
155    #[command(subcommand)]
156    Task(TaskCmd),
157    /// List notes.
158    Notes {
159        #[arg(long)]
160        scope: Option<String>,
161        #[arg(long)]
162        tag: Option<String>,
163    },
164    /// Operate on a single note.
165    #[command(subcommand)]
166    Note(NoteCmd),
167    /// Block until something happens on the bus (or the timeout passes).
168    Wait {
169        #[arg(long)]
170        timeout_seconds: Option<i64>,
171        /// Restrict to kinds: message, task, lock, note.
172        #[arg(long, value_delimiter = ',')]
173        kinds: Vec<String>,
174        /// Wake on every channel, not only the one this session works in.
175        #[arg(long)]
176        all_channels: bool,
177    },
178    /// Advisory locks on shared resources.
179    #[command(subcommand)]
180    Lock(LockCmd),
181    /// Summary of the team's recent activity.
182    Digest {
183        #[arg(long, default_value_t = 24)]
184        hours: i64,
185        /// Cover every channel, not only the one this session works in.
186        #[arg(long)]
187        all_channels: bool,
188    },
189    /// Escape hatch: call any tool with raw JSON arguments.
190    Call {
191        tool: String,
192        /// JSON object with the arguments, e.g. '{"key": "x"}'.
193        #[arg(long, default_value = "{}")]
194        args: String,
195    },
196}
197
198#[derive(Subcommand)]
199pub enum LockCmd {
200    Acquire {
201        name: String,
202        #[arg(long)]
203        ttl_seconds: Option<i64>,
204        #[arg(long)]
205        purpose: Option<String>,
206    },
207    Release {
208        name: String,
209    },
210    List,
211}
212
213#[derive(Subcommand)]
214pub enum TaskCmd {
215    Create {
216        key: String,
217        #[arg(long)]
218        title: String,
219        #[arg(long)]
220        description: Option<String>,
221        /// Comma-separated keys of tasks this one depends on.
222        #[arg(long, value_delimiter = ',')]
223        depends_on: Vec<String>,
224    },
225    Show {
226        key: String,
227    },
228    Claim {
229        key: String,
230        #[arg(long)]
231        lease_seconds: Option<i64>,
232    },
233    /// Claim the oldest available task, whatever it is.
234    Next {
235        #[arg(long)]
236        lease_seconds: Option<i64>,
237    },
238    Renew {
239        key: String,
240        #[arg(long)]
241        lease_seconds: Option<i64>,
242    },
243    Release {
244        key: String,
245    },
246    Done {
247        key: String,
248        #[arg(long)]
249        result: Option<String>,
250    },
251}
252
253#[derive(Subcommand)]
254pub enum NoteCmd {
255    Get {
256        key: String,
257        #[arg(long)]
258        scope: Option<String>,
259    },
260    Set {
261        key: String,
262        #[arg(long)]
263        value: String,
264        #[arg(long)]
265        scope: Option<String>,
266        #[arg(long, value_delimiter = ',')]
267        tags: Vec<String>,
268    },
269    Rm {
270        key: String,
271        #[arg(long)]
272        scope: Option<String>,
273    },
274    Search {
275        query: String,
276        #[arg(long)]
277        scope: Option<String>,
278    },
279}
280
281/// Strip nulls so optional flags the user did not pass are simply absent.
282pub mod mapping;
283pub mod render;
284
285use mapping::to_call;
286use render::render;
287
288pub async fn run(args: ClientArgs) -> anyhow::Result<()> {
289    let mut config = StreamableHttpClientTransportConfig::with_uri(args.url.clone());
290    config.auth_header = Some(args.token.clone());
291    config.allow_stateless = true;
292    if let Some(session) = args
293        .session
294        .as_deref()
295        .map(str::trim)
296        .filter(|s| !s.is_empty())
297    {
298        let value = session
299            .parse()
300            .with_context(|| format!("--session '{session}' is not a valid HTTP header value"))?;
301        config
302            .custom_headers
303            .insert(crate::auth::SESSION_HEADER.parse()?, value);
304    }
305    let transport = StreamableHttpClientTransport::from_config(config);
306
307    let client = ClientConfig::default()
308        .serve(transport)
309        .await
310        .context("could not connect to the bus (check --url and --token)")?;
311
312    let outcome = run_command(&client, &args).await;
313    let _ = client.cancel().await;
314    outcome
315}
316
317async fn run_command(
318    client: &rmcp::service::RunningService<rmcp::RoleClient, ClientConfig>,
319    args: &ClientArgs,
320) -> anyhow::Result<()> {
321    // `tools` is the one command that is not a tool call.
322    if matches!(args.command, ClientCmd::Tools) {
323        let tools = client.list_all_tools().await?;
324        if args.json {
325            println!("{}", serde_json::to_string_pretty(&tools)?);
326        } else {
327            for tool in tools {
328                println!(
329                    "{:<20} {}",
330                    tool.name,
331                    tool.description.as_deref().unwrap_or_default().trim()
332                );
333            }
334        }
335        return Ok(());
336    }
337
338    let (tool, call_args) = match &args.command {
339        ClientCmd::Call { tool, args: raw } => {
340            let parsed: Value = serde_json::from_str(raw)
341                .with_context(|| format!("--args is not valid JSON: {raw}"))?;
342            if !parsed.is_object() {
343                bail!("--args must be a JSON object");
344            }
345            (tool.clone(), parsed)
346        }
347        other => {
348            // `tools` is handled above and is the only command that maps to
349            // nothing; anything else reaching here without a mapping is a
350            // missing match arm, and says so instead of panicking.
351            let Some((tool, call_args)) = to_call(other)? else {
352                bail!("this subcommand has no MCP tool mapping yet");
353            };
354            (tool.to_string(), call_args)
355        }
356    };
357
358    let arguments: serde_json::Map<String, Value> =
359        serde_json::from_value(call_args).context("arguments did not form a JSON object")?;
360    let result = client
361        .call_tool(CallToolRequestParams::new(tool.clone()).with_arguments(arguments))
362        .await
363        .map_err(|e| anyhow::anyhow!("{tool} failed: {e}"))?;
364
365    if result.is_error == Some(true) {
366        bail!("{tool} returned an error: {:?}", result.content);
367    }
368    let value = result
369        .structured_content
370        .clone()
371        .unwrap_or_else(|| json!({ "ok": true }));
372
373    if args.json {
374        println!("{}", serde_json::to_string_pretty(&value)?);
375    } else {
376        render(&args.command, &value)?;
377    }
378    Ok(())
379}
ai_crew_sync/client/mod.rs

ai_crew_sync/client/
mod.rs
