Add Patina remote MCP connections

crates/buzz-acp/Cargo.toml

@@ -61,6 +61,7 @@ tracing-subscriber = { workspace = true }
 # Error handling
 thiserror = { workspace = true }
 anyhow = { workspace = true }
+zeroize = { workspace = true }
 
 # CLI
 clap = { version = "4", features = ["derive", "env"] }

crates/buzz-acp/src/acp.rs

@@ -22,16 +22,87 @@ const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB
 
 /// An MCP server configuration passed to `session/new`.
 ///
-/// Corresponds to the `McpServerStdio` variant in the ACP schema.
-/// All four fields are **required** by the schema (`args` and `env` may be empty arrays).
+/// ACP models stdio servers as an untagged object and HTTP servers as an
+/// object tagged with `type: "http"`, so the outer enum is untagged.
 #[derive(Debug, Clone, serde::Serialize)]
-pub struct McpServer {
+#[serde(untagged)]
+pub enum McpServer {
+    /// A child-process MCP server.
+    Stdio(McpServerStdio),
+    /// A remote Streamable HTTP MCP server.
+    Http(McpServerHttp),
+}
+
+impl McpServer {
+    /// Return the display name shared by both transport variants.
+    pub fn name(&self) -> &str {
+        match self {
+            Self::Stdio(server) => &server.name,
+            Self::Http(server) => &server.name,
+        }
+    }
+
+    /// Whether this server requires ACP HTTP MCP support.
+    pub fn is_http(&self) -> bool {
+        matches!(self, Self::Http(_))
+    }
+
+    /// Return the stdio payload when this is a child-process server.
+    #[cfg(test)]
+    pub fn as_stdio(&self) -> Option<&McpServerStdio> {
+        match self {
+            Self::Stdio(server) => Some(server),
+            Self::Http(_) => None,
+        }
+    }
+}
+
+/// ACP stdio MCP server.
+///
+/// All four fields are required by the schema (`args` and `env` may be empty
+/// arrays).
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct McpServerStdio {
     pub name: String,
     pub command: String,
     pub args: Vec<String>,
     pub env: Vec<EnvVar>,
 }
 
+/// ACP Streamable HTTP MCP server.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct McpServerHttp {
+    #[serde(rename = "type")]
+    pub transport: HttpTransport,
+    pub name: String,
+    pub url: String,
+    pub headers: Vec<HttpHeader>,
+}
+
+/// The only remote transport ACP currently defines.
+#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum HttpTransport {
+    Http,
+}
+
+/// A single HTTP header for a remote MCP server.
+#[derive(Clone, serde::Serialize, serde::Deserialize)]
+pub struct HttpHeader {
+    pub name: String,
+    pub value: String,
+}
+
+impl std::fmt::Debug for HttpHeader {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("HttpHeader")
+            .field("name", &self.name)
+            .field("value", &"[REDACTED]")
+            .finish()
+    }
+}
+
 /// A single environment variable for an MCP server.
 #[derive(Debug, Clone, serde::Serialize)]
 pub struct EnvVar {
@@ -121,6 +192,42 @@ fn agent_error_from_json(error: &serde_json::Value) -> AcpError {
     AcpError::AgentError { code, message }
 }
 
+/// Return a copy safe for logs and observer events.
+///
+/// `session/new` is the only ACP request that carries MCP credentials. Both
+/// stdio `env` values and HTTP `headers` values are replaced while names and
+/// the surrounding wire shape remain available for diagnosis.
+fn redact_acp_wire_message(message: &serde_json::Value) -> serde_json::Value {
+    let mut redacted = message.clone();
+    if redacted.get("method").and_then(serde_json::Value::as_str) != Some("session/new") {
+        return redacted;
+    }
+
+    let Some(servers) = redacted
+        .pointer_mut("/params/mcpServers")
+        .and_then(serde_json::Value::as_array_mut)
+    else {
+        return redacted;
+    };
+
+    for server in servers {
+        for key in ["env", "headers"] {
+            let Some(entries) = server
+                .get_mut(key)
+                .and_then(serde_json::Value::as_array_mut)
+            else {
+                continue;
+            };
+            for entry in entries {
+                if let Some(value) = entry.get_mut("value") {
+                    *value = serde_json::Value::String("[REDACTED]".to_string());
+                }
+            }
+        }
+    }
+    redacted
+}
+
 fn build_initialize_params() -> serde_json::Value {
     serde_json::json!({
         "protocolVersion": 2,
@@ -958,17 +1065,21 @@ impl AcpClient {
     /// (e.g., it's stuck or dead), the write would otherwise block forever.
     async fn write_ndjson(&mut self, value: &serde_json::Value) -> Result<(), AcpError> {
         const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
-        let line = serde_json::to_string(value)?;
-        tokio::time::timeout(WRITE_TIMEOUT, async {
+        use zeroize::Zeroize;
+
+        let mut line = serde_json::to_string(value)?;
+        let write_result = tokio::time::timeout(WRITE_TIMEOUT, async {
             self.stdin.write_all(line.as_bytes()).await?;
             self.stdin.write_all(b"\n").await?;
             self.stdin.flush().await?;
             Ok::<(), std::io::Error>(())
         })
-        .await
-        .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))?
-        .map_err(AcpError::Io)?;
-        self.observe("acp_write", value.clone());
+        .await;
+        line.zeroize();
+        write_result
+            .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))?
+            .map_err(AcpError::Io)?;
+        self.observe("acp_write", redact_acp_wire_message(value));
         Ok(())
     }
 
@@ -999,7 +1110,8 @@ impl AcpClient {
             "params": params,
         });
 
-        tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default());
+        let logged = redact_acp_wire_message(&msg);
+        tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&logged).unwrap_or_default());
 
         // Wrap write + read in a single timeout so a hung agent can't block forever.
         // We cannot use an async block that borrows `self` mutably across two awaits
@@ -2196,7 +2308,7 @@ mod tests {
     #[test]
     fn session_new_mcp_server_has_required_fields() {
         // Schema requires name, command, args, env — all present, args/env may be empty.
-        let server = McpServer {
+        let server = McpServer::Stdio(McpServerStdio {
             name: "test-mcp".into(),
             command: "/usr/local/bin/test-mcp-server".into(),
             args: vec![],
@@ -2210,7 +2322,7 @@ mod tests {
                     value: "nsec1abc".into(),
                 },
             ],
-        };
+        });
         let serialized = serde_json::to_value(&server).unwrap();
         assert_eq!(serialized["name"].as_str(), Some("test-mcp"));
         assert_eq!(
@@ -2227,6 +2339,69 @@ mod tests {
         );
     }
 
+    #[test]
+    fn session_new_http_mcp_server_matches_acp_wire_shape() {
+        let server = McpServer::Http(McpServerHttp {
+            transport: HttpTransport::Http,
+            name: "patina".into(),
+            url: "https://patina.so/mcp/acme".into(),
+            headers: vec![HttpHeader {
+                name: "Authorization".into(),
+                value: "Bearer pk_secret".into(),
+            }],
+        });
+
+        let serialized = serde_json::to_value(&server).unwrap();
+        assert_eq!(serialized["type"], "http");
+        assert_eq!(serialized["name"], "patina");
+        assert_eq!(serialized["url"], "https://patina.so/mcp/acme");
+        assert_eq!(serialized["headers"][0]["name"], "Authorization");
+        assert_eq!(serialized["headers"][0]["value"], "Bearer pk_secret");
+    }
+
+    #[test]
+    fn session_new_wire_observation_redacts_mcp_secrets() {
+        let message = serde_json::json!({
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "session/new",
+            "params": {
+                "cwd": "/tmp",
+                "mcpServers": [
+                    {
+                        "name": "buzz-mcp",
+                        "command": "buzz",
+                        "args": [],
+                        "env": [
+                            {"name": "BUZZ_PRIVATE_KEY", "value": "nsec_secret"}
+                        ]
+                    },
+                    {
+                        "type": "http",
+                        "name": "patina",
+                        "url": "https://patina.so/mcp/acme",
+                        "headers": [
+                            {"name": "Authorization", "value": "Bearer pk_secret"}
+                        ]
+                    }
+                ]
+            }
+        });
+
+        let redacted = redact_acp_wire_message(&message);
+        let rendered = redacted.to_string();
+        assert!(!rendered.contains("nsec_secret"));
+        assert!(!rendered.contains("pk_secret"));
+        assert_eq!(
+            redacted["params"]["mcpServers"][0]["env"][0]["value"],
+            "[REDACTED]"
+        );
+        assert_eq!(
+            redacted["params"]["mcpServers"][1]["headers"][0]["value"],
+            "[REDACTED]"
+        );
+    }
+
     #[test]
     fn session_prompt_request_format() {
         let prompt_text = "[Buzz @mention]\nChannel: test\nFrom: npub1...\nMessage: hello";

crates/buzz-acp/src/config.rs

@@ -12,9 +12,23 @@ use nostr::Keys;
 use thiserror::Error;
 use url::Url;
 use uuid::Uuid;
+use zeroize::Zeroize;
 
+use crate::acp::McpServerHttp;
 use crate::filter::SubscriptionRule;
 
+const MAX_REMOTE_MCP_STDIN_BYTES: usize = 64 * 1024;
+
+fn parse_remote_mcp_servers(input: &str) -> Result<Vec<McpServerHttp>, ConfigError> {
+    if input.len() > MAX_REMOTE_MCP_STDIN_BYTES {
+        return Err(ConfigError::ConfigFile(
+            "remote MCP stdin payload exceeds 64 KiB".into(),
+        ));
+    }
+    serde_json::from_str(input)
+        .map_err(|error| ConfigError::ConfigFile(format!("remote MCP stdin JSON: {error}")))
+}
+
 /// Default idle timeout (seconds) when neither `--idle-timeout` nor the
 /// deprecated `--turn-timeout` is set.
 ///
@@ -261,6 +275,13 @@ pub struct CliArgs {
     #[arg(long, env = "BUZZ_ACP_MCP_COMMAND", default_value = "")]
     pub mcp_command: String,
 
+    /// Read a JSON array of remote HTTP MCP servers from stdin at startup.
+    ///
+    /// Desktop uses this one-shot private pipe so bearer headers never enter
+    /// process arguments or environment variables.
+    #[arg(long, env = "BUZZ_ACP_REMOTE_MCP_STDIN", default_value_t = false)]
+    pub remote_mcp_stdin: bool,
+
     /// Idle timeout: max seconds of silence before killing a turn.
     /// Resets on any agent stdout activity.
     #[arg(long, env = "BUZZ_ACP_IDLE_TIMEOUT")]
@@ -495,6 +516,7 @@ pub struct Config {
     pub agent_command: String,
     pub agent_args: Vec<String>,
     pub mcp_command: String,
+    pub remote_mcp_servers: Vec<McpServerHttp>,
     pub idle_timeout_secs: u64,
     pub max_turn_duration_secs: u64,
     pub agents: u32,
@@ -1029,12 +1051,27 @@ impl Config {
 
         validate_multiple_event_handling(args.multiple_event_handling, args.dedup)?;
 
+        let remote_mcp_servers = if args.remote_mcp_stdin {
+            let mut input = String::new();
+            let mut reader = std::io::Read::take(
+                std::io::stdin().lock(),
+                (MAX_REMOTE_MCP_STDIN_BYTES + 1) as u64,
+            );
+            std::io::Read::read_to_string(&mut reader, &mut input)?;
+            let parsed = parse_remote_mcp_servers(&input);
+            input.zeroize();
+            parsed?
+        } else {
+            Vec::new()
+        };
+
         let config = Config {
             keys,
             relay_url: args.relay_url,
             agent_command,
             agent_args,
             mcp_command: args.mcp_command,
+            remote_mcp_servers,
             idle_timeout_secs,
             max_turn_duration_secs,
             agents: args.agents,
@@ -1413,6 +1450,7 @@ mod tests {
             agent_command: "goose".into(),
             agent_args: vec!["acp".into()],
             mcp_command: "".into(),
+            remote_mcp_servers: vec![],
             idle_timeout_secs: DEFAULT_IDLE_TIMEOUT_SECS,
             max_turn_duration_secs: DEFAULT_MAX_TURN_DURATION_SECS,
             agents: 1,
@@ -1451,6 +1489,27 @@ mod tests {
         }
     }
 
+    #[test]
+    fn remote_mcp_stdin_parser_accepts_http_server_shape() {
+        let servers = parse_remote_mcp_servers(
+            r#"[{"type":"http","name":"patina","url":"https://patina.so/mcp/acme","headers":[{"name":"Authorization","value":"Bearer pk_secret"}]}]"#,
+        )
+        .expect("valid ACP HTTP MCP payload should parse");
+
+        assert_eq!(servers.len(), 1);
+        assert_eq!(servers[0].name, "patina");
+        assert_eq!(servers[0].headers[0].name, "Authorization");
+    }
+
+    #[test]
+    fn remote_mcp_stdin_parser_rejects_invalid_and_oversized_payloads() {
+        assert!(parse_remote_mcp_servers("not-json").is_err());
+        let oversized = " ".repeat(MAX_REMOTE_MCP_STDIN_BYTES + 1);
+        let error = parse_remote_mcp_servers(&oversized)
+            .expect_err("oversized remote MCP payload must fail closed");
+        assert!(error.to_string().contains("64 KiB"));
+    }
+
     fn make_rule(
         name: &str,
         channels: ChannelScope,