feat(mcp): mesh_publish tool — surface publish_to / publish_to_direct (#56)

agent-mesh-cli/src/mcp.rs

@@ -195,6 +195,7 @@ async fn handle_request(
                 "mesh_whoami" => tool_whoami(state),
                 "mesh_peers" => tool_peers(state, &args).await,
                 "mesh_request" => tool_request(state, &args).await,
+                "mesh_publish" => tool_publish(state, &args).await,
                 other => Err(anyhow!("unknown tool: {other}")),
             }
             .map_err(RpcError::internal)?;
@@ -261,6 +262,35 @@ fn tool_definitions() -> Value {
                 },
                 "required": ["topic", "body"]
             }
+        },
+        {
+            "name": "mesh_publish",
+            "description": "Publish a body to a peer on a topic, fire-and-forget (no reply). The peer's bus fans it out to anyone subscribed to that topic. Name the peer the same two ways as mesh_request: `peer` (resolve by fingerprint over mDNS) OR `addr`+`pubkey` (direct dial, no discovery).",
+            "inputSchema": {
+                "type": "object",
+                "properties": {
+                    "peer": {
+                        "type": "string",
+                        "description": "Resolve mode: peer agent fingerprint — 64-char hex, or a unique prefix of a discovered peer. Omit when using addr+pubkey."
+                    },
+                    "addr": {
+                        "type": "string",
+                        "description": "Direct-dial socket address, e.g. `192.168.1.5:47800`. Requires `pubkey`; skips mDNS."
+                    },
+                    "pubkey": {
+                        "type": "string",
+                        "description": "Direct-dial peer agent ed25519 public key (64-char hex). Requires `addr`."
+                    },
+                    "topic": {
+                        "type": "string",
+                        "description": "Topic name (namespaced under our user fingerprint) to publish on."
+                    },
+                    "body": {
+                        "description": "Payload — JSON object (sent as-is) or string (sent as UTF-8)."
+                    }
+                },
+                "required": ["topic", "body"]
+            }
         }
     ])
 }
@@ -318,63 +348,26 @@ async fn tool_peers(state: &McpState, args: &Value) -> Result<String> {
 }
 
 async fn tool_request(state: &McpState, args: &Value) -> Result<String> {
-    // `peer` is required only in resolve mode; direct dial uses
-    // `addr`+`pubkey` instead (validated in the match below).
-    let peer_arg = args.get("peer").and_then(|p| p.as_str());
-    let topic_name = args
-        .get("topic")
-        .and_then(|t| t.as_str())
-        .ok_or_else(|| anyhow!("mesh_request needs `topic`"))?;
-    let body_val = args
-        .get("body")
-        .ok_or_else(|| anyhow!("mesh_request needs `body`"))?;
+    let (_topic_name, topic, body) = topic_and_body(state, args)?;
     let timeout_secs = args
         .get("timeout_secs")
         .and_then(Value::as_f64)
         .unwrap_or(30.0)
         .clamp(0.1, 600.0);
-
-    let body = match body_val {
-        Value::String(s) => s.clone().into_bytes(),
-        other => serde_json::to_vec(other)?,
-    };
-
-    let topic = Topic::new(state.bus.user_fingerprint(), topic_name);
     let timeout = Duration::from_secs_f64(timeout_secs);
 
-    // Two ways to name the peer, mutually exclusive:
-    //   * `addr` + `pubkey` — direct dial a known endpoint (no mDNS).
-    //   * `peer`            — resolve a fingerprint/prefix over mDNS.
-    let addr_arg = args.get("addr").and_then(Value::as_str);
-    let pubkey_arg = args.get("pubkey").and_then(Value::as_str);
+    let route = peer_route(state, args).await?;
+    let peer_fp = route.fingerprint();
 
     let started = std::time::Instant::now();
-    let (peer_fp, reply_bytes) = match (addr_arg, pubkey_arg) {
-        (Some(addr), Some(pubkey)) => {
-            let sock: SocketAddr = addr
-                .parse()
-                .with_context(|| format!("parse `addr` {addr:?} as <ip>:<port>"))?;
-            let endpoint = PeerEndpoint::new(parse_pubkey_hex(pubkey)?, sock);
-            let peer_fp = endpoint.fingerprint();
-            let reply = state
+    let reply_bytes = match route {
+        McpPeerRoute::Direct(endpoint) => {
+            state
                 .bus
                 .request_direct(endpoint, &topic, body, timeout)
-                .await?;
-            (peer_fp, reply)
-        }
-        (Some(_), None) | (None, Some(_)) => {
-            return Err(anyhow!(
-                "mesh_request direct dial needs both `addr` and `pubkey`"
-            ));
-        }
-        (None, None) => {
-            let peer_arg = peer_arg.ok_or_else(|| {
-                anyhow!("mesh_request needs `peer` (or `addr`+`pubkey` for direct dial)")
-            })?;
-            let peer_fp = resolve_peer(state, peer_arg).await?;
-            let reply = state.bus.request(peer_fp, &topic, body, timeout).await?;
-            (peer_fp, reply)
+                .await?
         }
+        McpPeerRoute::Resolve(peer_fp) => state.bus.request(peer_fp, &topic, body, timeout).await?,
     };
     let elapsed_ms = started.elapsed().as_millis() as u64;
 
@@ -388,6 +381,32 @@ async fn tool_request(state: &McpState, args: &Value) -> Result<String> {
     }))?)
 }
 
+async fn tool_publish(state: &McpState, args: &Value) -> Result<String> {
+    let (topic_name, topic, body) = topic_and_body(state, args)?;
+    let bytes = body.len();
+
+    let route = peer_route(state, args).await?;
+    let peer_fp = route.fingerprint();
+
+    // Fire-and-forget: the peer's bus fans the body out to its topic
+    // subscribers. No reply is awaited (unlike `mesh_request`).
+    match route {
+        McpPeerRoute::Direct(endpoint) => {
+            state.bus.publish_to_direct(endpoint, &topic, body).await?;
+        }
+        McpPeerRoute::Resolve(peer_fp) => {
+            state.bus.publish_to(peer_fp, &topic, body).await?;
+        }
+    }
+
+    Ok(serde_json::to_string_pretty(&json!({
+        "published": true,
+        "peer": peer_fp.hex(),
+        "topic": topic_name,
+        "bytes": bytes,
+    }))?)
+}
+
 /// Resolve a peer argument: full 64-char hex parses directly; anything
 /// shorter must be a unique prefix of a discovered peer's fingerprint.
 async fn resolve_peer(state: &McpState, arg: &str) -> Result<Fingerprint> {
@@ -426,6 +445,74 @@ fn parse_pubkey_hex(s: &str) -> Result<[u8; 32]> {
         .map_err(|v: Vec<u8>| anyhow!("`pubkey` decoded to {} bytes, need 32", v.len()))
 }
 
+/// Where a peer-addressed tool call is aimed: a directly-dialed endpoint
+/// or a fingerprint to resolve over mDNS.
+enum McpPeerRoute {
+    Direct(PeerEndpoint),
+    Resolve(Fingerprint),
+}
+
+impl McpPeerRoute {
+    /// The peer's agent fingerprint, however it was named.
+    fn fingerprint(&self) -> Fingerprint {
+        match self {
+            Self::Direct(ep) => ep.fingerprint(),
+            Self::Resolve(fp) => *fp,
+        }
+    }
+}
+
+/// Resolve the shared peer-naming arguments into a route. The two modes are
+/// mutually exclusive:
+///   * `addr` + `pubkey` — direct dial a known endpoint (no mDNS). Both
+///     are required together.
+///   * `peer`            — resolve a fingerprint/prefix over mDNS.
+///
+/// Shared by every peer-addressed tool (`mesh_request`, `mesh_publish`) so
+/// they name peers identically.
+async fn peer_route(state: &McpState, args: &Value) -> Result<McpPeerRoute> {
+    let addr_arg = args.get("addr").and_then(Value::as_str);
+    let pubkey_arg = args.get("pubkey").and_then(Value::as_str);
+    match (addr_arg, pubkey_arg) {
+        (Some(addr), Some(pubkey)) => {
+            let sock: SocketAddr = addr
+                .parse()
+                .with_context(|| format!("parse `addr` {addr:?} as <ip>:<port>"))?;
+            Ok(McpPeerRoute::Direct(PeerEndpoint::new(
+                parse_pubkey_hex(pubkey)?,
+                sock,
+            )))
+        }
+        (Some(_), None) | (None, Some(_)) => {
+            Err(anyhow!("direct dial needs both `addr` and `pubkey`"))
+        }
+        (None, None) => {
+            let peer_arg = args
+                .get("peer")
+                .and_then(Value::as_str)
+                .ok_or_else(|| anyhow!("needs `peer` (or `addr`+`pubkey` for direct dial)"))?;
+            Ok(McpPeerRoute::Resolve(resolve_peer(state, peer_arg).await?))
+        }
+    }
+}
+
+/// Extract the `topic` (namespaced under our user) and `body` bytes shared
+/// by `mesh_request` / `mesh_publish`. Returns the raw topic name too, for
+/// echoing in tool responses.
+fn topic_and_body(state: &McpState, args: &Value) -> Result<(String, Topic, Vec<u8>)> {
+    let topic_name = args
+        .get("topic")
+        .and_then(|t| t.as_str())
+        .ok_or_else(|| anyhow!("needs `topic`"))?;
+    let body_val = args.get("body").ok_or_else(|| anyhow!("needs `body`"))?;
+    let body = match body_val {
+        Value::String(s) => s.clone().into_bytes(),
+        other => serde_json::to_vec(other)?,
+    };
+    let topic = Topic::new(state.bus.user_fingerprint(), topic_name);
+    Ok((topic_name.to_string(), topic, body))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -467,7 +554,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread")]
-    async fn tools_list_names_all_three_tools() {
+    async fn tools_list_names_all_tools() {
         let state = quiet_state().await;
         let out = handle_request(&state, "tools/list", Value::Null)
             .await
@@ -478,7 +565,10 @@ mod tests {
             .iter()
             .map(|t| t["name"].as_str().unwrap())
             .collect();
-        assert_eq!(names, vec!["mesh_whoami", "mesh_peers", "mesh_request"]);
+        assert_eq!(
+            names,
+            vec!["mesh_whoami", "mesh_peers", "mesh_request", "mesh_publish"]
+        );
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -621,6 +711,45 @@ mod tests {
         assert!(err.message.contains("needs `peer`"), "got: {}", err.message);
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn publish_without_peer_or_addr_is_rejected() {
+        let state = quiet_state().await;
+        // mesh_publish shares the same peer-naming helper as mesh_request.
+        let err = handle_request(
+            &state,
+            "tools/call",
+            json!({
+                "name": "mesh_publish",
+                "arguments": { "topic": "t", "body": "x" }
+            }),
+        )
+        .await
+        .err()
+        .unwrap();
+        assert!(err.message.contains("needs `peer`"), "got: {}", err.message);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn publish_direct_needs_both_addr_and_pubkey() {
+        let state = quiet_state().await;
+        let err = handle_request(
+            &state,
+            "tools/call",
+            json!({
+                "name": "mesh_publish",
+                "arguments": { "addr": "127.0.0.1:47800", "topic": "t", "body": "x" }
+            }),
+        )
+        .await
+        .err()
+        .unwrap();
+        assert!(
+            err.message.contains("both `addr` and `pubkey`"),
+            "got: {}",
+            err.message
+        );
+    }
+
     #[test]
     fn parse_pubkey_hex_roundtrips_and_rejects_bad_length() {
         assert_eq!(parse_pubkey_hex(&"cd".repeat(32)).unwrap(), [0xcd; 32]);

agent-mesh-cli/tests/mcp_stdio.rs

@@ -189,6 +189,78 @@ async fn mcp_server_direct_addr_round_trips_to_live_responder() {
     responder.close().await.expect("close responder");
 }
 
+/// Deterministic `mesh_publish` fan-out: `amesh mcp` publishes a body to a
+/// quiet-bound in-process subscriber `Bus` by explicit `addr`+`pubkey` over
+/// real QUIC on loopback (no mDNS), and the subscriber's topic
+/// `broadcast::Receiver` receives it (#56). Fire-and-forget — no reply,
+/// unlike the request round-trip above.
+#[tokio::test(flavor = "multi_thread")]
+async fn mcp_server_direct_addr_publishes_to_subscriber() {
+    // 1. keygen + start the MCP server + handshake.
+    let home = tempfile::tempdir().unwrap();
+    let status = Command::new(env!("CARGO_BIN_EXE_amesh"))
+        .arg("--home")
+        .arg(home.path())
+        .arg("keygen")
+        .status()
+        .await
+        .expect("run keygen");
+    assert!(status.success(), "keygen must succeed");
+    let user = UserKey::load(&home.path().join("user.key")).expect("load generated key");
+
+    let mut client = McpClient::spawn(home.path());
+    let init = client.call("initialize", json!({})).await;
+    assert_eq!(init["result"]["serverInfo"]["name"], "amesh-mcp");
+    client.notify("notifications/initialized").await;
+
+    // 2. Bring up a quiet-bound subscriber and subscribe BEFORE publishing
+    //    so the broadcast channel exists to buffer the message.
+    let subscriber_agent = echo_agent(&user);
+    let subscriber_pubkey = subscriber_agent.public_bytes();
+    let subscriber = Bus::bind_with(&user, subscriber_agent, 0, BusOptions { announce: false })
+        .await
+        .expect("bind subscriber");
+    let subscriber_port = subscriber.local_port();
+    let topic = Topic::new(user.fingerprint(), "notes/v1");
+    let mut sub_rx = subscriber.subscribe(&topic).await;
+
+    // 3. Publish via explicit addr+pubkey — no mesh_peers, no mDNS.
+    let publish = client
+        .call(
+            "tools/call",
+            json!({
+                "name": "mesh_publish",
+                "arguments": {
+                    "addr": format!("127.0.0.1:{subscriber_port}"),
+                    "pubkey": hex::encode(subscriber_pubkey),
+                    "topic": "notes/v1",
+                    "body": { "note": "hello sub" }
+                }
+            }),
+        )
+        .await;
+    let publish = McpClient::tool_text(&publish);
+    assert_eq!(publish["published"], true, "publish ack; got {publish}");
+    assert_eq!(publish["topic"], "notes/v1");
+
+    // 4. The subscriber's broadcast receiver gets the published body.
+    let got = tokio::time::timeout(STEP_TIMEOUT, sub_rx.recv())
+        .await
+        .expect("subscriber must receive the publish before timeout")
+        .expect("broadcast recv");
+    let got: Value = serde_json::from_slice(&got).expect("published body is JSON");
+    assert_eq!(got, json!({ "note": "hello sub" }));
+
+    // 5. Clean shutdown.
+    drop(client.stdin);
+    let status = tokio::time::timeout(STEP_TIMEOUT, client.child.wait())
+        .await
+        .expect("server must exit after stdin closes")
+        .expect("wait");
+    assert!(status.success(), "clean exit, got {status:?}");
+    subscriber.close().await.expect("close subscriber");
+}
+
 // Real mDNS multicast discovery (steps 5–6 resolve the responder by
 // fingerprint over multicast); flaky on hosted CI. See the module doc.
 // The deterministic per-PR coverage of the same round-trip is
@@ -226,7 +298,10 @@ async fn mcp_server_round_trips_request_to_live_responder() {
         .iter()
         .map(|t| t["name"].as_str().unwrap())
         .collect();
-    assert_eq!(names, vec!["mesh_whoami", "mesh_peers", "mesh_request"]);
+    assert_eq!(
+        names,
+        vec!["mesh_whoami", "mesh_peers", "mesh_request", "mesh_publish"]
+    );
 
     let whoami = client
         .call(