feat: Phase 2 — OAuth2 refresh, SMTP transport, persistent MCP state

crates/Cargo.toml

@@ -36,3 +36,4 @@ url = "2"
 anyhow = "1"
 tempfile = "3"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+lettre = { version = "0.11", default-features = false, features = ["tokio1-rustls-tls", "smtp-transport", "builder"] }

crates/connector-hub/src/main.rs

@@ -257,6 +257,7 @@ fn build_catalogue() -> anyhow::Result<hub_core::Catalogue> {
                             json_schema: raw.parameters,
                         },
                         tags: raw.tags,
+                        transport: hub_core::Transport::Http,
                     });
                 }
 
@@ -269,5 +270,66 @@ fn build_catalogue() -> anyhow::Result<hub_core::Catalogue> {
         }
     }
 
+    register_email_operations(&mut catalogue);
+
     Ok(catalogue)
 }
+
+fn register_email_operations(catalogue: &mut hub_core::Catalogue) {
+    catalogue.register(hub_core::Operation {
+        id: hub_core::OperationId("email.send".into()),
+        provider: "email".into(),
+        summary: "Send an email via SMTP".into(),
+        description: "Send an email message using the configured SMTP transport. Supports plain text, HTML, and multipart bodies. Requires SMTP credentials configured via EMAIL_SMTP_* environment variables.".into(),
+        mutation_class: hub_policy::MutationClass::Mutating,
+        http_method: "POST".into(),
+        path_template: String::new(),
+        parameters: hub_core::ParameterSchema {
+            json_schema: serde_json::json!({
+                "type": "object",
+                "required": ["to", "subject"],
+                "properties": {
+                    "to": {
+                        "description": "Recipient email address(es). String or array of strings.",
+                        "oneOf": [
+                            {"type": "string"},
+                            {"type": "array", "items": {"type": "string"}}
+                        ]
+                    },
+                    "cc": {
+                        "description": "CC recipients",
+                        "oneOf": [
+                            {"type": "string"},
+                            {"type": "array", "items": {"type": "string"}}
+                        ]
+                    },
+                    "bcc": {
+                        "description": "BCC recipients",
+                        "oneOf": [
+                            {"type": "string"},
+                            {"type": "array", "items": {"type": "string"}}
+                        ]
+                    },
+                    "subject": {
+                        "type": "string",
+                        "description": "Email subject"
+                    },
+                    "body": {
+                        "type": "string",
+                        "description": "Plain text body"
+                    },
+                    "body_html": {
+                        "type": "string",
+                        "description": "HTML body (sent as multipart/alternative with plain text)"
+                    },
+                    "from": {
+                        "type": "string",
+                        "description": "Override sender address (defaults to configured from_address)"
+                    }
+                }
+            }),
+        },
+        tags: vec!["email".into(), "smtp".into()],
+        transport: hub_core::Transport::Smtp,
+    });
+}

crates/connector-hub/src/mcp.rs

@@ -1,10 +1,17 @@
+use std::sync::Arc;
+
 use rmcp::handler::server::wrapper::Parameters;
 use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo};
 use rmcp::{ErrorData, ServiceExt, schemars, tool, tool_handler, tool_router};
 use serde::Deserialize;
 
 #[derive(Clone)]
-pub struct HubMcpServer;
+pub struct HubMcpServer {
+    dispatcher: Arc<hub_core::Dispatcher>,
+    auth: Arc<hub_auth::AuthStore>,
+    policy: Arc<hub_policy::Policy>,
+    net: Arc<hub_net::NetClient>,
+}
 
 #[derive(Deserialize, schemars::JsonSchema)]
 struct SearchParams {
@@ -41,8 +48,7 @@ impl HubMcpServer {
         description = "List all available providers and their operation counts"
     )]
     async fn list_providers(&self) -> Result<String, ErrorData> {
-        let catalogue =
-            super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
+        let catalogue = self.dispatcher.catalogue();
 
         let mut providers = Vec::new();
         for name in catalogue.providers() {
@@ -68,8 +74,7 @@ impl HubMcpServer {
         &self,
         Parameters(params): Parameters<SearchParams>,
     ) -> Result<String, ErrorData> {
-        let catalogue =
-            super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
+        let catalogue = self.dispatcher.catalogue();
 
         let results = catalogue.search(&params.query, params.provider.as_deref());
         let items: Vec<_> = results
@@ -96,8 +101,7 @@ impl HubMcpServer {
         &self,
         Parameters(params): Parameters<DescribeParams>,
     ) -> Result<CallToolResult, ErrorData> {
-        let catalogue =
-            super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
+        let catalogue = self.dispatcher.catalogue();
 
         let op_id = hub_core::OperationId(params.id.clone());
         match catalogue.get(&op_id) {
@@ -119,39 +123,19 @@ impl HubMcpServer {
         &self,
         Parameters(params): Parameters<CallParams>,
     ) -> Result<CallToolResult, ErrorData> {
-        let catalogue =
-            super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
-
         let op_id = hub_core::OperationId(params.id.clone());
-        let op = catalogue.get(&op_id).ok_or_else(|| {
-            ErrorData::internal_error(format!("Unknown operation: {}", params.id), None)
-        })?;
-
-        if params.dry_run.unwrap_or(false) {
-            let outcome = hub_core::ExecutionOutcome::DryRun {
-                would_execute: op.summary.clone(),
-                mutation_class: format!("{:?}", op.mutation_class),
-            };
-            return Ok(CallToolResult::success(vec![ContentBlock::text(
-                serde_json::to_string_pretty(&outcome).unwrap(),
-            )]));
-        }
 
-        let policy = hub_policy::Policy::deny_all();
-        let auth = hub_auth::AuthStore::from_env();
-        let net = hub_net::NetClient::new(hub_net::SsrfPolicy::default());
-
-        let dispatcher = hub_core::Dispatcher::new(catalogue);
-        let result = dispatcher
+        let result = self
+            .dispatcher
             .call(
                 &op_id,
                 params.args,
                 params.account.as_deref(),
-                false,
+                params.dry_run.unwrap_or(false),
                 params.confirmation_token.as_deref(),
-                &policy,
-                &auth,
-                &net,
+                &self.policy,
+                &self.auth,
+                &self.net,
             )
             .await;
 
@@ -167,7 +151,14 @@ impl HubMcpServer {
 
     #[tool(name = "health", description = "Check the health of the connector hub")]
     async fn health(&self) -> String {
-        serde_json::json!({"ok": true, "service": "connector-hub"}).to_string()
+        let catalogue = self.dispatcher.catalogue();
+        serde_json::json!({
+            "ok": true,
+            "service": "connector-hub",
+            "providers": catalogue.providers().len(),
+            "operations": catalogue.len(),
+        })
+        .to_string()
     }
 }
 
@@ -183,9 +174,25 @@ impl rmcp::ServerHandler for HubMcpServer {
 pub async fn serve() -> anyhow::Result<()> {
     tracing::info!("starting MCP stdio server");
 
-    let server = HubMcpServer;
-    let transport = rmcp::transport::io::stdio();
+    let catalogue = super::build_catalogue()?;
+    let dispatcher = hub_core::Dispatcher::new(catalogue);
+
+    let policy_path = std::path::Path::new("permissions.toml");
+    let policy = if policy_path.exists() {
+        let content = std::fs::read_to_string(policy_path)?;
+        hub_policy::Policy::from_toml(&content)?
+    } else {
+        hub_policy::Policy::deny_all()
+    };
 
+    let server = HubMcpServer {
+        dispatcher: Arc::new(dispatcher),
+        auth: Arc::new(hub_auth::AuthStore::from_env()),
+        policy: Arc::new(policy),
+        net: Arc::new(hub_net::NetClient::new(hub_net::SsrfPolicy::default())),
+    };
+
+    let transport = rmcp::transport::io::stdio();
     let running = server.serve(transport).await?;
     tracing::info!("MCP server running on stdio");
 

crates/hub-auth/Cargo.toml

@@ -10,3 +10,5 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
+tokio = { workspace = true }
+reqwest = { workspace = true }

crates/hub-auth/src/credential.rs

@@ -45,4 +45,13 @@ pub enum AuthMethod {
         password: String,
         token_url: String,
     },
+    #[serde(rename = "smtp")]
+    Smtp {
+        host: String,
+        port: u16,
+        username: String,
+        password: String,
+        tls: bool,
+        from_address: String,
+    },
 }