Merge pull request #97 from XpressAI/fix/agent-apps-tools-session

fix: agent apps, MCP tool routing, and session continuity

Author: wmeddie

Cargo.lock

@@ -37,6 +37,11 @@ croner = "2"
 bytes = "1"
 sysinfo = "0.34"
 notify = "7"
+lettre = { version = "0.11", features = ["tokio1-native-tls", "builder"] }
+async-imap = "0.10"
+async-native-tls = "0.5"
+async-std = { version = "1", features = ["default"] }
+base64 = "0.22"
 
 # Local LLM (optional, enabled by default via "local-llm" feature)
 llama-cpp-2 = { workspace = true, optional = true }

crates/xpressclaw-core/Cargo.toml

@@ -132,13 +132,18 @@ impl HarnessClient {
 
     /// Send a message to the agent's persistent session.
     /// Returns a stream of SSE chunks (may be buffered by the SDK).
+    ///
+    /// `history` is the full conversation so far (excluding the current message).
+    /// The harness injects it when starting a fresh session so the agent has context
+    /// even after a container restart.
     pub async fn send_session_message(
         &self,
         message: &str,
         conversation_id: &str,
         sender_name: &str,
         sender_type: &str,
         system_prompt: &str,
+        history: &serde_json::Value,
     ) -> Result<ChatStream> {
         let url = format!("{}/v1/session/send", self.base_url);
 
@@ -148,6 +153,7 @@ impl HarnessClient {
             "sender_name": sender_name,
             "sender_type": sender_type,
             "system_prompt": system_prompt,
+            "history": history,
             "stream": true,
         });
 

crates/xpressclaw-core/src/agents/harness.rs

@@ -52,13 +52,17 @@ pub fn deliver(db: &Arc<Database>, connector_name: &str, channel_name: &str, mes
         "telegram" => deliver_telegram(&connector.config, &channel.config, message),
         "webhook" => deliver_webhook(&connector.config, &channel.config, message),
         "file_watcher" => deliver_file(&channel.config, message),
+        "slack" => deliver_slack(&connector.config, &channel.config, message),
+        "github" => deliver_github(&connector.config, &channel.config, message),
+        "jira" => deliver_jira(&connector.config, &channel.config, message),
+        "email" => deliver_email(&connector.config, &channel.config, message),
         other => {
             info!(
                 connector_type = other,
                 connector = connector_name,
                 channel = channel_name,
                 message,
-                "sink delivery (stub connector — message logged only)"
+                "sink delivery (unknown connector — message logged only)"
             );
             Ok(())
         }
@@ -163,3 +167,278 @@ fn deliver_file(channel_config: &Value, message: &str) -> Result<()> {
 
     Ok(())
 }
+
+/// Send via Slack chat.postMessage API.
+fn deliver_slack(connector_config: &Value, channel_config: &Value, message: &str) -> Result<()> {
+    let bot_token = connector_config
+        .get("bot_token")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("slack: missing bot_token".into()))?
+        .to_string();
+
+    let channel_id = channel_config
+        .get("channel_id")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("slack: missing channel_id in channel config".into()))?
+        .to_string();
+
+    let message = message.to_string();
+    tokio::spawn(async move {
+        let body = json!({
+            "channel": channel_id,
+            "text": message,
+        });
+        let client = reqwest::Client::new();
+        if let Err(e) = client
+            .post("https://slack.com/api/chat.postMessage")
+            .header("Authorization", format!("Bearer {}", bot_token))
+            .header("Content-Type", "application/json; charset=utf-8")
+            .json(&body)
+            .send()
+            .await
+        {
+            error!(error = %e, "slack chat.postMessage failed");
+        }
+    });
+
+    Ok(())
+}
+
+/// Send via GitHub API (comment on issue or create issue).
+fn deliver_github(connector_config: &Value, _channel_config: &Value, message: &str) -> Result<()> {
+    let token = connector_config
+        .get("token")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("github: missing token".into()))?
+        .to_string();
+
+    let owner = connector_config
+        .get("owner")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("github: missing owner".into()))?
+        .to_string();
+
+    let repo = connector_config
+        .get("repo")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("github: missing repo".into()))?
+        .to_string();
+
+    // Try to parse message as JSON to extract issue_number for comments;
+    // otherwise create a new issue with the message as body.
+    let message = message.to_string();
+    tokio::spawn(async move {
+        let client = reqwest::Client::builder()
+            .user_agent("xpressclaw/0.1")
+            .build()
+            .unwrap_or_else(|_| reqwest::Client::new());
+
+        // Attempt to parse JSON to check for issue_number
+        let parsed: Option<Value> = serde_json::from_str(&message).ok();
+        let issue_number = parsed
+            .as_ref()
+            .and_then(|v| v.get("issue_number"))
+            .and_then(|v| v.as_u64());
+
+        if let Some(num) = issue_number {
+            let url = format!(
+                "https://api.github.com/repos/{}/{}/issues/{}/comments",
+                owner, repo, num
+            );
+            let body_text = parsed
+                .as_ref()
+                .and_then(|v| v.get("body"))
+                .and_then(|v| v.as_str())
+                .unwrap_or(&message);
+            let body = json!({ "body": body_text });
+            if let Err(e) = client
+                .post(&url)
+                .header("Authorization", format!("token {}", token))
+                .header("Accept", "application/vnd.github.v3+json")
+                .json(&body)
+                .send()
+                .await
+            {
+                error!(error = %e, "github comment POST failed");
+            }
+        } else {
+            let url = format!("https://api.github.com/repos/{}/{}/issues", owner, repo);
+            let body = json!({
+                "title": "New issue from xpressclaw",
+                "body": message,
+            });
+            if let Err(e) = client
+                .post(&url)
+                .header("Authorization", format!("token {}", token))
+                .header("Accept", "application/vnd.github.v3+json")
+                .json(&body)
+                .send()
+                .await
+            {
+                error!(error = %e, "github create issue failed");
+            }
+        }
+    });
+
+    Ok(())
+}
+
+/// Send via Jira API (add comment to issue).
+fn deliver_jira(connector_config: &Value, _channel_config: &Value, message: &str) -> Result<()> {
+    use base64::Engine as _;
+
+    let base_url = connector_config
+        .get("base_url")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("jira: missing base_url".into()))?
+        .trim_end_matches('/')
+        .to_string();
+
+    let email = connector_config
+        .get("email")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("jira: missing email".into()))?
+        .to_string();
+
+    let api_token = connector_config
+        .get("api_token")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("jira: missing api_token".into()))?
+        .to_string();
+
+    // Try to extract issue key from JSON message
+    let parsed: Option<Value> = serde_json::from_str(message).ok();
+    let issue_key = parsed
+        .as_ref()
+        .and_then(|v| v.get("issue_key").or(v.get("key")))
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| {
+            Error::Connector("jira: message must contain issue_key or key field".into())
+        })?
+        .to_string();
+
+    let body_text = parsed
+        .as_ref()
+        .and_then(|v| v.get("body"))
+        .and_then(|v| v.as_str())
+        .unwrap_or(message)
+        .to_string();
+
+    tokio::spawn(async move {
+        let credentials = format!("{}:{}", email, api_token);
+        let encoded = base64::engine::general_purpose::STANDARD.encode(credentials.as_bytes());
+        let auth = format!("Basic {}", encoded);
+
+        let url = format!("{}/rest/api/3/issue/{}/comment", base_url, issue_key);
+        let body = json!({
+            "body": {
+                "type": "doc",
+                "version": 1,
+                "content": [{
+                    "type": "paragraph",
+                    "content": [{
+                        "type": "text",
+                        "text": body_text
+                    }]
+                }]
+            }
+        });
+
+        let client = reqwest::Client::new();
+        if let Err(e) = client
+            .post(&url)
+            .header("Authorization", &auth)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+        {
+            error!(error = %e, "jira comment POST failed");
+        }
+    });
+
+    Ok(())
+}
+
+/// Send via SMTP email.
+fn deliver_email(connector_config: &Value, channel_config: &Value, message: &str) -> Result<()> {
+    let smtp_host = connector_config
+        .get("smtp_host")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("email: missing smtp_host".into()))?
+        .to_string();
+
+    let smtp_port = connector_config
+        .get("smtp_port")
+        .and_then(|v| v.as_u64())
+        .unwrap_or(587) as u16;
+
+    let username = connector_config
+        .get("username")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("email: missing username".into()))?
+        .to_string();
+
+    let password = connector_config
+        .get("password")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("email: missing password".into()))?
+        .to_string();
+
+    let to_addr = channel_config
+        .get("to")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::Connector("email: missing 'to' in channel config".into()))?
+        .to_string();
+
+    let message = message.to_string();
+    let from = username.clone();
+
+    tokio::spawn(async move {
+        use lettre::message::header::ContentType;
+        use lettre::transport::smtp::authentication::Credentials;
+        use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};
+
+        let email = match Message::builder()
+            .from(match from.parse() {
+                Ok(addr) => addr,
+                Err(e) => {
+                    error!(error = %e, "invalid from address for email delivery");
+                    return;
+                }
+            })
+            .to(match to_addr.parse() {
+                Ok(addr) => addr,
+                Err(e) => {
+                    error!(error = %e, "invalid to address for email delivery");
+                    return;
+                }
+            })
+            .subject("Message from xpressclaw")
+            .header(ContentType::TEXT_PLAIN)
+            .body(message)
+        {
+            Ok(e) => e,
+            Err(e) => {
+                error!(error = %e, "failed to build email");
+                return;
+            }
+        };
+
+        let creds = Credentials::new(username, password);
+
+        let mailer = match AsyncSmtpTransport::<Tokio1Executor>::starttls_relay(&smtp_host) {
+            Ok(builder) => builder.port(smtp_port).credentials(creds).build(),
+            Err(e) => {
+                error!(error = %e, "failed to create SMTP transport");
+                return;
+            }
+        };
+
+        if let Err(e) = mailer.send(email).await {
+            error!(error = %e, "SMTP send failed");
+        }
+    });
+
+    Ok(())
+}