//! MCP tool surface for `hive-matrix-daemon`, served directly over //! streamable-http — no stdio bridge, no per-turn respawn, no //! round-trip unix socket. The daemon already owns the account //! registry in-process (built at startup from the restored matrix-sdk //! `Client`s), so each tool call resolves `account` against //! [`crate::accounts::Registry`] and calls straight into //! [`crate::handlers`]. Mirrors `hive-bash-mcp::mcp`'s shape (and, one //! level up, `hive-agent-mcp::mcp::serve_http`): a persistent daemon, //! stable URL claude reconnects to every turn instead of respawning a //! stdio child. //! //! Multi-account: every tool carries an optional `account` arg naming //! which matrix account to act as (a `name` from //! `hyperhive.matrixAccounts`); omitting it selects the agent's primary //! account (or errors, listing the choices, when more than one account //! is configured — see [`crate::accounts::Registry::resolve`]). use std::sync::Arc; use rmcp::{ ServerHandler, handler::server::wrapper::Parameters, schemars::{self, JsonSchema}, tool, tool_handler, tool_router, }; use serde::Deserialize; use crate::accounts::Registry; use crate::handlers; use crate::protocol::{DaemonResponse, InviteAction}; /// Turn a `DaemonResponse` into the string claude sees as the tool /// result. Ok payloads are pretty-printed JSON; errors get a clear /// "matrix error: …" prefix so claude can pattern-match on it. fn render(resp: DaemonResponse) -> String { match resp { DaemonResponse::Ok { payload } => { serde_json::to_string_pretty(&payload).unwrap_or_else(|_| payload.to_string()) } DaemonResponse::Error { message } => format!("matrix error: {message}"), } } #[derive(Debug, Deserialize, JsonSchema)] struct SendMessageArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`). /// Aliases are server-resolved at send time. room: String, /// Message body. Markdown is rendered to HTML by the daemon /// (`text_markdown`); plain text passes through unchanged. body: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct SendDmArgs { /// Matrix user id of the recipient (`@user:server`). DM room is /// created if one doesn't already exist between this agent and /// the recipient. user_id: String, body: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct SendFileArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`). room: String, /// Absolute path to a local file readable by the agent (e.g. a built /// PDF under the agent's workspace). MIME type is inferred from the /// extension. 50 MiB cap. path: String, /// Optional caption, sent as a follow-up text message in the room. #[serde(default)] caption: Option, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct OpenDmArgs { /// Matrix user id (`@user:server`) to open a DM with. The DM room is /// created if one doesn't already exist. user_id: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct SendReactionArgs { room: String, /// Full matrix event id of the message being reacted to. event_id: String, /// Reaction key — usually an emoji (`👍`, `❤️`) but any string /// works per matrix spec. key: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct SendReplyArgs { room: String, event_id: String, body: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct MarkReadArgs { room: String, event_id: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct SendRedactArgs { room: String, /// Full matrix event id of the message to redact. event_id: String, /// Optional human-readable reason recorded on the redaction event. #[serde(default)] reason: Option, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct ListRoomsArgs { /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct ListRoomMembersArgs { room: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct ReadRoomArgs { room: String, /// Maximum events to return (default 50, max 200). Newest first. #[serde(default)] limit: Option, /// Anchor at this event id and read *forward* from it: returns the anchor /// plus the `limit - 1` events after it (chronologically newer). Use to /// continue reading from a known event. Mutually exclusive with `until`. #[serde(default)] from: Option, /// Anchor at this event id and read *backward* to it: returns the anchor /// plus the `limit - 1` events before it (into the past). Use to page /// older context around a known event (e.g. the target of a reply). /// Mutually exclusive with `from`. #[serde(default)] until: Option, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct DownloadFileArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`). room: String, /// Event id of the attachment message (from `read_room`, where media /// events show a `[file: …]` / `[image: …]` marker). event_id: String, /// Optional absolute destination path. Omit to write a temp file /// named after the attachment; the returned `path` is where to read it. #[serde(default)] dest_path: Option, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct ListInvitesArgs { /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct JoinRoomArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`). room: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct ResolveInviteArgs { /// Matrix room id (`!abc:server`) or alias (`#name:server`) you have /// a pending invite to. room: String, /// What to do with the invite: `"accept"` (join the room) or /// `"reject"` (decline and leave). action: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct InviteUserArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`) /// to invite the user into. You must already be a member. room: String, /// Matrix user id of the invitee (`@user:server`). user_id: String, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] account: Option, } #[derive(Clone)] struct MatrixMcp { registry: Arc, } impl MatrixMcp { /// Resolve `account` against the registry, rendering the "unknown /// account" / "ambiguous, pick one" error the same way a handler /// error would render (so a bad `account` arg and a bad room/event /// arg look the same to claude). fn resolve(&self, account: Option<&str>) -> Result<&matrix_sdk::Client, String> { self.registry .resolve(account) .map(std::convert::AsRef::as_ref) } } #[tool_router] impl MatrixMcp { #[tool( description = "Post a plain-text or markdown message to a matrix room. \ `room` is either a room id (!abc:server) or alias (#name:server). \ Returns the new event id. Rejected with a hint if the room still has \ unread messages — read_room then mark_read the latest event first so \ you don't talk over messages you haven't seen." )] async fn send_message(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::send_message(client, &args.room, &args.body).await) } #[tool(description = "Open (or reuse) a direct message room with `user_id` \ (@user:server) and post `body` to it. If the DM room already exists \ and has unread messages, the send is rejected with a hint — read_room \ then mark_read the latest event first.")] async fn send_dm(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::send_dm(client, &args.user_id, &args.body).await) } #[tool( description = "Upload a local file and post it as an attachment to a matrix \ room. `room` is a room id (!abc:server) or alias (#name:server); `path` \ is an absolute path to a local file (MIME inferred from extension, 50 MiB \ cap); optional `caption` is sent as a follow-up message. Rejected if the \ room has unread messages — read_room then mark_read first." )] async fn send_file(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::send_file(client, &args.room, &args.path, args.caption.as_deref()).await) } #[tool(description = "Resolve (find-or-create) the DM room with `user_id` \ (@user:server) and return its room id, without sending anything. Use the \ returned room id with the room-based tools (`send_file`, `send_message`, \ …) to deliver into the DM — there is no per-tool DM variant.")] async fn open_dm(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::open_dm(client, &args.user_id).await) } #[tool( description = "React to a specific matrix event with an emoji or short \ string `key`. Matrix-spec annotation; renders as a reaction in \ standard clients." )] async fn send_reaction(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::send_reaction(client, &args.room, &args.event_id, &args.key).await) } #[tool(description = "Reply to a specific matrix event in a room, threaded \ via m.in_reply_to. Returns the reply's event id. Rejected with a hint \ if the room still has unread messages — read_room then mark_read the \ latest event first.")] async fn send_reply(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::send_reply(client, &args.room, &args.event_id, &args.body).await) } #[tool(description = "Mark a specific event as read for this agent. Updates \ the room's unread indicator + sends a read receipt other \ participants can see.")] async fn mark_read(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::mark_read(client, &args.room, &args.event_id).await) } #[tool(description = "Redact (delete) a specific matrix event in a room — \ asks the homeserver to strip the event's content, optionally with a \ `reason`. Works on your own events; redacting others' needs moderator \ power level. Irreversible.")] async fn send_redact(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render( handlers::send_redact(client, &args.room, &args.event_id, args.reason.as_deref()).await, ) } #[tool( description = "List rooms this agent has joined. Each row has the room \ id, canonical alias (when set), display name, and joined-member count." )] async fn list_rooms(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::list_rooms(client).await) } #[tool( description = "List rooms this agent has been invited to but not yet joined. \ Each row has the room id, canonical alias (when set), and display name. \ Use `resolve_invite` to accept or reject an invite." )] async fn list_invites(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::list_invites(client)) } #[tool( description = "Join a matrix room by id (!abc:server) or alias (#name:server). \ Also accepts a pending invite if one exists, but for invites prefer \ `resolve_invite` (which can also reject). Use `join_room` to join a \ public room you weren't invited to. After joining the room appears in \ `list_rooms`." )] async fn join_room(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::join_room(client, &args.room).await) } #[tool( description = "Accept or reject a pending matrix invite. `room` is a room id \ (!abc:server) or alias (#name:server) you were invited to; `action` is \ \"accept\" (join the room) or \"reject\" (decline and leave). See pending \ invites with `list_invites`." )] async fn resolve_invite(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; let action = match args.action.trim().to_ascii_lowercase().as_str() { "accept" => InviteAction::Accept, "reject" => InviteAction::Reject, other => { return format!( "matrix error: invalid action {other:?} — use \"accept\" or \"reject\"" ); } }; render(handlers::resolve_invite(client, &args.room, action).await) } #[tool( description = "Invite a user to a matrix room you're already in. `room` is a \ room id (!abc:server) or alias (#name:server); `user_id` is the invitee \ (@user:server). You must have a high enough power level in the room to \ invite. The invitee then sees a pending invite they accept with `join_room`." )] async fn invite_user(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::invite_user(client, &args.room, &args.user_id).await) } #[tool( description = "List the members of a matrix room (joined-state only). \ Each row carries the user id and resolved display name." )] async fn list_room_members(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::list_room_members(client, &args.room).await) } #[tool(description = "Read events from a matrix room (default 50, max 200), \ newest first. Returns each event's id, sender, timestamp, type, \ best-effort plain-text body, and the id of the event it replies to \ (when any). By default returns the most recent events. Pass `from` \ = an event id to read forward from that event (anchor + newer \ events), or `until` = an event id to read backward to it (anchor + \ older events) — e.g. to fetch the context around a reply target \ outside the current window. `from` and `until` are mutually \ exclusive.")] async fn read_room(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render(handlers::read_room(client, &args.room, args.limit, args.from, args.until).await) } #[tool( description = "Download a media attachment from a matrix message to a local \ file and return its path. `room` is a room id/alias; `event_id` is the \ attachment message (read_room shows media as `[file: …]`/`[image: …]`); \ optional `dest_path` overrides the temp destination. The read-side \ counterpart of send_file." )] async fn download_file(&self, Parameters(args): Parameters) -> String { let client = match self.resolve(args.account.as_deref()) { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; render( handlers::download_file( client, &args.room, &args.event_id, args.dest_path.as_deref(), ) .await, ) } } #[tool_handler(instructions = "Matrix client for an agent on a hyperhive swarm. Use \ `send_message` to post in a joined room, `send_dm` to message a \ specific user, `send_reaction` to react with an emoji, `send_reply` \ to thread a reply, `mark_read` to acknowledge an event, `send_redact` \ to delete an event. Discover \ rooms with `list_rooms`, members with `list_room_members`, recent \ timeline with `read_room`. See pending invites with `list_invites`; \ accept or reject an invite with `resolve_invite`; join a public \ room with `join_room`. Room references accept ids (!abc:server) \ or aliases (#name:server); user references use @user:server. Every \ tool takes an optional `account` (a name from the agent's matrix \ accounts) — omit it to act as the primary account. \ `send_file` uploads a local file and posts it as a room attachment \ (MIME inferred from extension, 50 MiB cap); `send_file` is rejected \ with a hint if the room has unread messages — read_room then mark_read \ first. `download_file` downloads an attachment from a message event \ to a local file and returns its path; pair with `read_room` which \ surfaces attachments as `[file: name]`, `[image: name]`, \ `[audio: name]`, `[video: name]` markers. \ IMPORTANT: `send_message`, `send_dm`, `send_file`, `send_reply` are \ all rejected if the room has unread messages — always call `read_room` \ then `mark_read` on the latest event before sending to a room you \ haven't read yet.")] impl ServerHandler for MatrixMcp {} /// Plain-JSON status endpoint for the primary account's unread rooms — /// NOT part of the claude-facing MCP tool surface. `hive-agent-mcp`'s /// `get_loose_ends` hits this directly (same container, loopback only) /// to prepend a matrix-unread entry, mirroring the pre-http unix-socket /// `unread_summary` side channel the daemon used to serve. Best-effort: /// only resolves when exactly one account is configured (same rule as /// an MCP tool call omitting `account`) — a multi-account agent's extra /// accounts aren't reachable from here, same restriction the caller /// already documents for cross-agent queries. async fn unread_summary_handler( axum::extract::State(registry): axum::extract::State>, ) -> axum::Json { let payload = match registry.resolve(None) { Ok(client) => handlers::unread_summary(client.as_ref()).await, Err(message) => crate::protocol::DaemonResponse::error(message), }; axum::Json( serde_json::to_value(payload) .unwrap_or_else(|e| serde_json::json!({ "kind": "error", "message": e.to_string() })), ) } /// Run the MCP server over HTTP (rmcp streamable-http transport) on /// `addr`, dispatching against `registry`. Also serves a small /// non-MCP `/unread-summary` status endpoint (see /// `unread_summary_handler`). /// /// Sole transport — there is no stdio mode. Long-lived so claude /// reconnects to the stable URL each turn instead of respawning a /// stdio child; since the daemon already owns the account registry /// in-process, tool calls need no round-trip to anywhere. /// /// Binds loopback only in practice; the default `allowed_hosts` /// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere /// else for `/mcp`; `/unread-summary` is plain axum with no such /// guard, but the listener itself is loopback-only so this is moot. /// /// # Errors /// /// Returns an error if the listener cannot bind `addr` or the HTTP /// server exits with a fatal error. pub async fn serve_http(addr: std::net::SocketAddr, registry: Arc) -> anyhow::Result<()> { use rmcp::transport::streamable_http_server::{ StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, }; let session_manager = std::sync::Arc::new(LocalSessionManager::default()); let mcp_registry = registry.clone(); let service = StreamableHttpService::new( move || { Ok(MatrixMcp { registry: mcp_registry.clone(), }) }, session_manager, StreamableHttpServerConfig::default(), ); let app = axum::Router::new() .nest_service("/mcp", service) .route( "/unread-summary", axum::routing::get(unread_summary_handler), ) .with_state(registry); let listener = tokio::net::TcpListener::bind(addr).await?; tracing::info!(%addr, "serving hive-matrix MCP over streamable-http at /mcp"); axum::serve(listener, app).await?; Ok(()) }