crabtalk-runtime 0.0.19

Crabtalk agent runtime — tool dispatch, MCP, skills, and memory
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

//! HTTP transport — POST JSON-RPC with SSE response support.

use anyhow::{Context, Result, bail};

pub struct HttpTransport {
    client: reqwest::Client,
    url: String,
    session_id: Option<String>,
}

impl HttpTransport {
    pub fn new(url: &str) -> Self {
        Self {
            client: reqwest::Client::new(),
            url: url.to_string(),
            session_id: None,
        }
    }

    pub async fn request(&mut self, msg: serde_json::Value) -> Result<serde_json::Value> {
        let mut req = self
            .client
            .post(self.url.as_str())
            .header("Content-Type", "application/json")
            .header("Accept", "application/json, text/event-stream");
        if let Some(sid) = self.session_id.as_deref() {
            req = req.header("Mcp-Session-Id", sid);
        }
        let resp = req.json(&msg).send().await.context("HTTP request failed")?;

        let status = resp.status();
        if let Some(sid) = resp.headers().get("mcp-session-id") {
            self.session_id = sid.to_str().ok().map(String::from);
        }

        let content_type = resp
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("")
            .to_string();

        let body = resp.text().await.context("failed to read response body")?;

        if !status.is_success() {
            bail!("HTTP {status}: {body}");
        }

        if content_type.contains("text/event-stream") {
            parse_sse_response(&body)
        } else {
            serde_json::from_str(&body).context("invalid JSON response")
        }
    }

    pub async fn notify(&mut self, msg: serde_json::Value) -> Result<()> {
        let mut req = self
            .client
            .post(self.url.as_str())
            .header("Content-Type", "application/json");
        if let Some(sid) = self.session_id.as_deref() {
            req = req.header("Mcp-Session-Id", sid);
        }
        req.json(&msg).send().await.context("HTTP notify failed")?;
        Ok(())
    }
}

/// Extract the last JSON-RPC message from an SSE response body.
/// Takes only the final `data:` line — intermediate messages (progress
/// notifications, etc.) are intentionally skipped since we only use
/// request/response methods, not streaming notifications.
fn parse_sse_response(body: &str) -> Result<serde_json::Value> {
    let mut last_data = None;
    for line in body.lines() {
        if let Some(data) = line.strip_prefix("data:") {
            let data = data.trim();
            if !data.is_empty() {
                last_data = Some(data);
            }
        }
    }
    let data = last_data.context("no data in SSE response")?;
    serde_json::from_str(data).context("invalid JSON in SSE data")
}