hyperhive/hive-matrix-mcp/src/bin/mcp.rs

//! `hive-matrix-mcp` binary — stdio MCP server claude spawns per turn.
//! Thin protocol bridge: every tool call → connect to the daemon's
//! unix socket → write a JSON request line → read the JSON response →
//! return the payload (or the error message) to claude.
//!
//! No matrix-sdk dep at this entrypoint. The daemon owns the heavy
//! Client + sync; the bridge is pure serde + tokio I/O. Means the MCP
//! binary cold-starts in milliseconds even though the daemon takes
//! seconds to bring up sync.
//!
//! 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. The bridge wraps each operation in a [`DaemonRequest`]
//! envelope carrying that account; the daemon routes to the matching
//! client.

use anyhow::{Context, Result};
use rmcp::{
    ServerHandler, ServiceExt,
    handler::server::wrapper::Parameters,
    schemars::{self, JsonSchema},
    tool, tool_handler, tool_router,
    transport::stdio,
};
use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;

use hive_matrix_mcp::paths;
use hive_matrix_mcp::protocol::{DaemonOp, DaemonRequest, DaemonResponse, InviteAction};

/// Send `req` to the daemon and read back the response. Each call is a
/// fresh unix-socket connection — short-lived (the daemon dispatch is
/// a single round-trip) so connection pooling would be over-engineering.
async fn round_trip(req: DaemonRequest) -> Result<DaemonResponse> {
    let socket = paths::daemon_socket();
    let stream = UnixStream::connect(&socket).await.with_context(|| {
        format!(
            "matrix daemon unreachable at {} — it may be starting up or restarting \
             (the daemon rebinds its socket a few seconds after a restart); retry shortly",
            socket.display()
        )
    })?;
    let (reader, mut writer) = stream.into_split();
    let mut line = serde_json::to_string(&req)?;
    line.push('\n');
    writer
        .write_all(line.as_bytes())
        .await
        .context("write request to daemon socket")?;
    writer.shutdown().await.ok();
    let mut buf = String::new();
    BufReader::new(reader)
        .read_line(&mut buf)
        .await
        .context("read response from daemon socket")?;
    serde_json::from_str(&buf).context("parse daemon response")
}

/// Wrap an op in a [`DaemonRequest`] envelope for `account` and send it.
async fn call(account: Option<String>, op: DaemonOp) -> Result<DaemonResponse> {
    round_trip(DaemonRequest { account, op }).await
}

/// 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: Result<DaemonResponse>) -> String {
    match resp {
        Ok(DaemonResponse::Ok { payload }) => {
            serde_json::to_string_pretty(&payload).unwrap_or_else(|_| payload.to_string())
        }
        Ok(DaemonResponse::Error { message }) => format!("matrix error: {message}"),
        Err(e) => format!("matrix bridge error: {e:#}"),
    }
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
    /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
    /// Omit to use the agent's primary account.
    #[serde(default)]
    account: Option<String>,
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
    /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
    /// Omit to use the agent's primary account.
    #[serde(default)]
    account: Option<String>,
}

#[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<String>,
}

#[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<String>,
}

#[derive(Debug, Deserialize, JsonSchema)]
struct ReadRoomArgs {
    room: String,
    /// Maximum events to return (default 50, max 200). Newest first.
    #[serde(default)]
    limit: Option<usize>,
    /// 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<String>,
    /// 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<String>,
    /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
    /// Omit to use the agent's primary account.
    #[serde(default)]
    account: Option<String>,
}

#[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<String>,
    /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
    /// Omit to use the agent's primary account.
    #[serde(default)]
    account: Option<String>,
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
}

#[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<String>,
}

struct MatrixBridge {
    #[allow(
        dead_code,
        reason = "populated by the #[tool_router] macro; the generated \
                  ServerHandler wiring consumes it, the field is never read directly"
    )]
    tool_router: rmcp::handler::server::router::tool::ToolRouter<Self>,
}

impl MatrixBridge {
    fn new() -> Self {
        Self {
            tool_router: Self::tool_router(),
        }
    }
}

#[tool_router]
impl MatrixBridge {
    #[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<SendMessageArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendMessage {
                    room: args.room,
                    body: 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<SendDmArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendDm {
                    user_id: args.user_id,
                    body: 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<SendFileArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendFile {
                    room: args.room,
                    path: args.path,
                    caption: args.caption,
                },
            )
            .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<OpenDmArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::OpenDm {
                    user_id: 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<SendReactionArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendReaction {
                    room: args.room,
                    event_id: args.event_id,
                    key: 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<SendReplyArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendReply {
                    room: args.room,
                    event_id: args.event_id,
                    body: 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<MarkReadArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::MarkRead {
                    room: args.room,
                    event_id: 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<SendRedactArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::SendRedact {
                    room: args.room,
                    event_id: args.event_id,
                    reason: args.reason,
                },
            )
            .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<ListRoomsArgs>) -> String {
        render(call(args.account, DaemonOp::ListRooms).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<ListInvitesArgs>) -> String {
        render(call(args.account, DaemonOp::ListInvites).await)
    }

    #[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<JoinRoomArgs>) -> String {
        render(call(args.account, DaemonOp::JoinRoom { room: 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<ResolveInviteArgs>) -> String {
        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(
            call(
                args.account,
                DaemonOp::ResolveInvite {
                    room: 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<InviteUserArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::InviteUser {
                    room: args.room,
                    user_id: 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<ListRoomMembersArgs>) -> String {
        render(call(args.account, DaemonOp::ListRoomMembers { room: 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<ReadRoomArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::ReadRoom {
                    room: args.room,
                    limit: args.limit,
                    from: args.from,
                    until: 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<DownloadFileArgs>) -> String {
        render(
            call(
                args.account,
                DaemonOp::DownloadFile {
                    room: args.room,
                    event_id: args.event_id,
                    dest_path: args.dest_path,
                },
            )
            .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 MatrixBridge {}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")),
        )
        .with_writer(std::io::stderr)
        .init();

    // Standalone-degraded boot: matrix isn't provisioned for this agent
    // (no token file) → exit 0 cleanly so claude doesn't register a
    // matrix MCP server it can never use.
    //
    // Gate on the TOKEN, not the daemon socket. The daemon binds its
    // socket only AFTER restoring its matrix session (~10s on a cold
    // boot), so an exists-check on the socket here raced the daemon's
    // startup: during that window the socket was absent, the bridge
    // exited, and claude lost the matrix tools for the WHOLE session
    // (the bridge isn't respawned mid-turn). The token, by contrast, is
    // written by hive-c0re at provisioning time and is present well
    // before the daemon finishes booting — so it cleanly distinguishes
    // "matrix not set up for this agent" (token absent → exit) from
    // "daemon still coming up" (token present → keep serving). When the
    // token exists we serve regardless of socket state: tool calls
    // `connect()` per-call and simply error until the daemon is up, but
    // the tools stay registered for the session.
    let token_file = paths::token_file();
    if !tokio::fs::try_exists(&token_file).await.unwrap_or(false) {
        tracing::warn!(
            path = %token_file.display(),
            "matrix not provisioned (no token file); exiting cleanly so MCP startup doesn't fail"
        );
        return Ok(());
    }

    let bridge = MatrixBridge::new();
    let service = bridge
        .serve(stdio())
        .await
        .context("serve MCP over stdio")?;
    service
        .waiting()
        .await
        .context("MCP service exited unexpectedly")?;
    Ok(())
}