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

//! 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<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>,
}

#[derive(Clone)]
struct MatrixMcp {
    registry: Arc<Registry>,
}

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<SendMessageArgs>) -> 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<SendDmArgs>) -> 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<SendFileArgs>) -> 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<OpenDmArgs>) -> 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<SendReactionArgs>) -> 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<SendReplyArgs>) -> 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<MarkReadArgs>) -> 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<SendRedactArgs>) -> 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<ListRoomsArgs>) -> 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<ListInvitesArgs>) -> 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<JoinRoomArgs>) -> 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<ResolveInviteArgs>) -> 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<InviteUserArgs>) -> 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<ListRoomMembersArgs>) -> 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<ReadRoomArgs>) -> 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<DownloadFileArgs>) -> 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<Arc<Registry>>,
) -> axum::Json<serde_json::Value> {
    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<Registry>) -> 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(())
}