ai_crew_sync/tools/
mod.rs

1pub mod events;
2pub mod locks;
3pub mod messaging;
4pub mod notes;
5pub mod presence;
6pub mod tasks;
7
8use rmcp::{
9    ErrorData, Json, ServerHandler,
10    handler::server::{router::tool::ToolRouter, wrapper::Parameters},
11    model::{ServerCapabilities, ServerInfo},
12    service::RequestContext,
13    tool, tool_handler, tool_router,
14};
15use schemars::JsonSchema;
16use serde::Deserialize;
17use sqlx::PgPool;
18
19use crate::{
20    auth::AuthCtx,
21    events::EventHub,
22    model::{DigestResult, WhoAmI},
23    store,
24};
25
26pub const INSTRUCTIONS: &str = r#"
27Shared coordination bus for a team of AI coding agents. Every agent in the
28team is connected to the same bus, so anything you write here is visible to your
29teammates' agents, and anything they write is visible to you.
30
31Identity is taken from your bearer token; you never pass your own name as an
32argument. Call `whoami` once at the start of a session to learn your handle and
33see whether anything is waiting for you.
34
35Four capabilities:
36
37- Messaging. `post_message` to a `channel` (broadcast to the whole team) or to a
38  single agent via `to` (direct message). `read_messages` returns only what you
39  have not seen yet by default, and advances your read cursor.
40- Task coordination. Before starting shared work, `claim_task` so two agents do
41  not do the same job twice. Claims hold a lease that expires, so call
42  `renew_task_lease` on long jobs and `complete_task` or `release_task` when you
43  stop. An expired lease can be taken over by anyone. Tasks can depend on other
44  tasks (`depends_on` at creation); blocked tasks cannot be claimed until their
45  dependencies are done, and `claim_next_task` skips them.
46- Locks. `acquire_lock` before touching a contended resource ("deploy:staging",
47  "schema:users"); `release_lock` when finished. Lighter than a task, expires on
48  its own.
49- Presence. `heartbeat` publishes what you are working on (repo, branch, a short
50  activity string). `list_agents` shows who else is active right now.
51- Shared notes. `set_note` / `get_note` / `search_notes` are the team's durable
52  memory: decisions, gotchas, deploy state. Prefer a note over repeating the
53  same explanation in chat.
54- Waiting. When you are blocked on teammates, call `wait_for_updates` instead of
55  polling: it blocks until a relevant message/task/lock/note event arrives or
56  the timeout passes. `team_digest` summarises the last hours of team activity —
57  useful at session start to catch up.
58- Asking. When you need an answer from a specific teammate to continue,
59  `ask_agent` sends them the question and waits for the reply in one call. If
60  you receive a question (a direct message marked `"question": true`), answer
61  promptly with `post_message` (`to` the asker, `reply_to` the question id) —
62  their agent is blocked waiting on you.
63
64Conventions worth following: keep messages short and factual, scope notes by
65repository name, and use task keys that a human would recognise.
66"#;
67
68/// The MCP server. Cheap to clone: `PgPool` and `EventHub` are `Arc`s
69/// internally, and the tool router is rebuilt per session by the transport's
70/// service factory.
71#[derive(Clone)]
72pub struct Bus {
73    pub db: PgPool,
74    pub hub: EventHub,
75    pub tool_router: ToolRouter<Self>,
76}
77
78impl Bus {
79    pub fn new(db: PgPool, hub: EventHub) -> Self {
80        let tool_router = Self::core_router()
81            + Self::messaging_router()
82            + Self::tasks_router()
83            + Self::presence_router()
84            + Self::notes_router()
85            + Self::locks_router()
86            + Self::events_router();
87        Self {
88            db,
89            hub,
90            tool_router,
91        }
92    }
93}
94
95/// Pull the authenticated identity out of the HTTP request that carried this
96/// tool call. The bearer middleware put it there; if it is missing, something
97/// is routing around authentication and we refuse rather than guess.
98pub fn auth_of(ctx: &RequestContext<rmcp::RoleServer>) -> Result<AuthCtx, ErrorData> {
99    ctx.extensions
100        .get::<http::request::Parts>()
101        .and_then(|parts| parts.extensions.get::<AuthCtx>())
102        .cloned()
103        .ok_or_else(|| {
104            ErrorData::invalid_request(
105                "no authentication context on this request; the server is misconfigured",
106                None,
107            )
108        })
109}
110
111#[tool_router(router = core_router, vis = "pub")]
112impl Bus {
113    #[tool(
114        description = "Identify yourself on the bus: your agent handle, your team, \
115                       how many unread direct messages you have and how many tasks \
116                       you currently hold. Call this first in a session."
117    )]
118    async fn whoami(
119        &self,
120        ctx: RequestContext<rmcp::RoleServer>,
121    ) -> Result<Json<WhoAmI>, ErrorData> {
122        let auth = auth_of(&ctx)?;
123        Ok(Json(store::whoami(&self.db, &auth).await?))
124    }
125
126    #[tool(
127        description = "Summarise the team's last hours: channel activity, tasks that moved, \
128                       notes touched, who was around, active locks. Call it at session start \
129                       to catch up, or to prepare a standup. Direct messages are excluded."
130    )]
131    async fn team_digest(
132        &self,
133        ctx: RequestContext<rmcp::RoleServer>,
134        Parameters(args): Parameters<DigestArgs>,
135    ) -> Result<Json<DigestResult>, ErrorData> {
136        let auth = auth_of(&ctx)?;
137        Ok(Json(
138            store::digest::team_digest(&self.db, &auth, args.hours.unwrap_or(24)).await?,
139        ))
140    }
141}
142
143#[derive(Debug, Deserialize, JsonSchema)]
144pub struct DigestArgs {
145    /// Window to summarise, in hours (1-336). Defaults to 24.
146    #[serde(default)]
147    pub hours: Option<i64>,
148}
149
150#[tool_handler(router = self.tool_router)]
151impl ServerHandler for Bus {
152    fn get_info(&self) -> ServerInfo {
153        let mut info = ServerInfo::new(ServerCapabilities::builder().enable_tools().build());
154        info.instructions = Some(INSTRUCTIONS.trim().to_string());
155        info
156    }
157}
ai_crew_sync/tools/mod.rs

ai_crew_sync/tools/
mod.rs
