crow_memory_sdk/
lib.rs

1//! crow-memory-sdk — reqwest client for the crow-memory HTTP server.
2//!
3//! Same method surface as the old in-process MemoryStore, minus open/path:
4//! `MemoryClient::connect(url)`. Connection failures and 502/503/504 retry
5//! with exponential backoff (v1 lesson: a service blip with no backoff
6//! killed the whole experiment). 4xx and 500 fail fast — 500s from the
7//! store are not retried so non-idempotent writes never double-apply.
8
9use std::time::Duration;
10
11use anyhow::Context;
12use crow_memory_types::{
13    AddMessageRequest, AddMessageResponse, CreateAgentRequest, ErrorResponse, LookupPromptRequest,
14    LookupPromptResponse, MaxAgentIdxResponse, SearchMessagesRequest,
15};
16use reqwest::Method;
17use serde::{de::DeserializeOwned, Serialize};
18
19pub use crow_memory_types::{
20    AgentRecord, MessageRecord, PromptRecord, SessionInfo, DEFAULT_MEMORY_PORT,
21};
22
23const MAX_RETRIES: u32 = 5;
24const BASE_BACKOFF_MS: u64 = 100;
25const MAX_BACKOFF_MS: u64 = 2000;
26
27/// Default crow-memory server URL. Honors `CROW_MEMORY_PORT` (set it in
28/// `{config_dir}/.env` or container env — docker-compose friendly), else
29/// `DEFAULT_MEMORY_PORT` (27697 — CROWS on a phone keypad).
30pub fn default_memory_url() -> String {
31    let port = std::env::var("CROW_MEMORY_PORT")
32        .ok()
33        .and_then(|s| s.parse::<u16>().ok())
34        .unwrap_or(DEFAULT_MEMORY_PORT);
35    format!("http://127.0.0.1:{port}")
36}
37
38pub struct MemoryClient {
39    http: reqwest::Client,
40    base_url: String,
41}
42
43impl MemoryClient {
44    /// Lazy connect — no I/O until the first call.
45    /// `base_url` e.g. `http://127.0.0.1:27697`.
46    pub fn connect(base_url: impl Into<String>) -> Self {
47        Self {
48            http: reqwest::Client::new(),
49            base_url: base_url.into().trim_end_matches('/').to_string(),
50        }
51    }
52
53    pub fn base_url(&self) -> &str {
54        &self.base_url
55    }
56
57    pub async fn health(&self) -> anyhow::Result<()> {
58        let _: serde_json::Value = self.send(Method::GET, "/healthz", &[], None::<&()>).await?;
59        Ok(())
60    }
61
62    // ---- prompts ----
63
64    pub async fn lookup_or_create_prompt(
65        &self,
66        template: &str,
67        name: &str,
68    ) -> anyhow::Result<String> {
69        let r: LookupPromptResponse = self
70            .send(
71                Method::POST,
72                "/v1/prompts/lookup",
73                &[],
74                Some(&LookupPromptRequest {
75                    template: template.to_string(),
76                    name: name.to_string(),
77                }),
78            )
79            .await?;
80        Ok(r.prompt_id)
81    }
82
83    pub async fn get_prompt(&self, prompt_id: &str) -> anyhow::Result<Option<PromptRecord>> {
84        self.send_opt(Method::GET, &format!("/v1/prompts/{prompt_id}"), &[])
85            .await
86    }
87
88    // ---- agents ----
89
90    #[allow(clippy::too_many_arguments)]
91    pub async fn create_agent(
92        &self,
93        agent_id: &str,
94        session_id: &str,
95        agent_idx: i64,
96        cwd: &str,
97        prompt_id: &str,
98        prompt_args: &serde_json::Value,
99        system_prompt: &str,
100        tool_definitions: &serde_json::Value,
101        request_params: &serde_json::Value,
102        model_identifier: &str,
103    ) -> anyhow::Result<()> {
104        let resp = self
105            .send_raw(
106                Method::POST,
107                "/v1/agents",
108                &[],
109                Some(&CreateAgentRequest {
110                    agent_id: agent_id.to_string(),
111                    session_id: session_id.to_string(),
112                    agent_idx,
113                    cwd: cwd.to_string(),
114                    prompt_id: prompt_id.to_string(),
115                    prompt_args: prompt_args.clone(),
116                    system_prompt: system_prompt.to_string(),
117                    tool_definitions: tool_definitions.clone(),
118                    request_params: request_params.clone(),
119                    model_identifier: model_identifier.to_string(),
120                }),
121            )
122            .await?;
123        check_success(resp).await.map(|_| ())
124    }
125
126    pub async fn get_agent(&self, agent_id: &str) -> anyhow::Result<Option<AgentRecord>> {
127        self.send_opt(Method::GET, &format!("/v1/agents/{agent_id}"), &[])
128            .await
129    }
130
131    pub async fn list_agents(
132        &self,
133        session_id: Option<&str>,
134    ) -> anyhow::Result<Vec<AgentRecord>> {
135        let mut query: Vec<(&str, String)> = Vec::new();
136        if let Some(s) = session_id {
137            query.push(("session_id", s.to_string()));
138        }
139        self.send(Method::GET, "/v1/agents", &query, None::<&()>).await
140    }
141
142    pub async fn get_max_agent_idx(&self, session_id: &str) -> anyhow::Result<i64> {
143        let r: MaxAgentIdxResponse = self
144            .send(
145                Method::GET,
146                "/v1/max-agent-idx",
147                &[("session_id", session_id.to_string())],
148                None::<&()>,
149            )
150            .await?;
151        Ok(r.max_idx)
152    }
153
154    // ---- messages ----
155
156    pub async fn add_message(
157        &self,
158        agent_id: &str,
159        message: &serde_json::Value,
160        usage: Option<&serde_json::Value>,
161    ) -> anyhow::Result<i64> {
162        let r: AddMessageResponse = self
163            .send(
164                Method::POST,
165                "/v1/messages",
166                &[],
167                Some(&AddMessageRequest {
168                    agent_id: agent_id.to_string(),
169                    message: message.clone(),
170                    usage: usage.cloned(),
171                }),
172            )
173            .await?;
174        Ok(r.id)
175    }
176
177    pub async fn load_messages(
178        &self,
179        agent_id: &str,
180    ) -> anyhow::Result<Vec<serde_json::Value>> {
181        self.send(
182            Method::GET,
183            &format!("/v1/agents/{agent_id}/messages"),
184            &[],
185            None::<&()>,
186        )
187        .await
188    }
189
190    pub async fn query_messages_by_agent(
191        &self,
192        agent_id: &str,
193        order_asc: bool,
194        limit: usize,
195        role: Option<&str>,
196    ) -> anyhow::Result<Vec<MessageRecord>> {
197        let mut query: Vec<(&str, String)> = vec![
198            ("order_asc", order_asc.to_string()),
199            ("limit", limit.to_string()),
200        ];
201        if let Some(r) = role {
202            query.push(("role", r.to_string()));
203        }
204        self.send(
205            Method::GET,
206            &format!("/v1/agents/{agent_id}/messages/query"),
207            &query,
208            None::<&()>,
209        )
210        .await
211    }
212
213    pub async fn search_messages(
214        &self,
215        query: &str,
216        limit: usize,
217        role: Option<&str>,
218    ) -> anyhow::Result<Vec<MessageRecord>> {
219        self.send(
220            Method::POST,
221            "/v1/messages/search",
222            &[],
223            Some(&SearchMessagesRequest {
224                query: query.to_string(),
225                limit,
226                role: role.map(str::to_string),
227            }),
228        )
229        .await
230    }
231
232    // ---- sessions ----
233
234    pub async fn list_sessions(
235        &self,
236        limit: usize,
237        offset: usize,
238    ) -> anyhow::Result<Vec<SessionInfo>> {
239        self.send(
240            Method::GET,
241            "/v1/sessions",
242            &[
243                ("limit", limit.to_string()),
244                ("offset", offset.to_string()),
245            ],
246            None::<&()>,
247        )
248        .await
249    }
250
251    pub async fn get_sessions_by_cwd(&self, cwd: &str) -> anyhow::Result<Vec<SessionInfo>> {
252        self.send(
253            Method::GET,
254            "/v1/sessions/by-cwd",
255            &[("cwd", cwd.to_string())],
256            None::<&()>,
257        )
258        .await
259    }
260
261    // ---- plumbing ----
262
263    async fn send<B: Serialize, R: DeserializeOwned>(
264        &self,
265        method: Method,
266        path: &str,
267        query: &[(&str, String)],
268        body: Option<&B>,
269    ) -> anyhow::Result<R> {
270        let resp = self.send_raw(method, path, query, body).await?;
271        parse_body(resp).await
272    }
273
274    /// GET that maps 404 → None.
275    async fn send_opt<R: DeserializeOwned>(
276        &self,
277        method: Method,
278        path: &str,
279        query: &[(&str, String)],
280    ) -> anyhow::Result<Option<R>> {
281        let resp = self
282            .send_raw::<()>(method, path, query, None)
283            .await?;
284        if resp.status() == reqwest::StatusCode::NOT_FOUND {
285            return Ok(None);
286        }
287        Ok(Some(parse_body(resp).await?))
288    }
289
290    /// Request with retry: connect errors + 502/503/504 back off
291    /// exponentially; everything else returns immediately.
292    async fn send_raw<B: Serialize>(
293        &self,
294        method: Method,
295        path: &str,
296        query: &[(&str, String)],
297        body: Option<&B>,
298    ) -> anyhow::Result<reqwest::Response> {
299        let url = format!("{}{}", self.base_url, path);
300        let mut attempt: u32 = 0;
301        loop {
302            let mut req = self.http.request(method.clone(), &url);
303            if !query.is_empty() {
304                req = req.query(query);
305            }
306            if let Some(b) = body {
307                req = req.json(b);
308            }
309            let retryable = match req.send().await {
310                Ok(resp) => {
311                    let st = resp.status();
312                    if matches!(
313                        st,
314                        reqwest::StatusCode::BAD_GATEWAY
315                            | reqwest::StatusCode::SERVICE_UNAVAILABLE
316                            | reqwest::StatusCode::GATEWAY_TIMEOUT
317                    ) {
318                        Some(format!("HTTP {st}"))
319                    } else {
320                        return Ok(resp);
321                    }
322                }
323                Err(e) if e.is_connect() => Some(format!("{e}")),
324                Err(e) => {
325                    return Err(anyhow::Error::new(e))
326                        .context(format!("memory server request failed: {url}"))
327                }
328            };
329            if attempt >= MAX_RETRIES {
330                anyhow::bail!(
331                    "memory server unreachable after {} retries: {} ({url})",
332                    MAX_RETRIES,
333                    retryable.unwrap_or_default()
334                );
335            }
336            attempt += 1;
337            let backoff = (BASE_BACKOFF_MS.saturating_mul(1 << attempt)).min(MAX_BACKOFF_MS);
338            tracing::warn!(
339                "memory server unavailable ({}); retry {attempt}/{MAX_RETRIES} in {backoff}ms",
340                retryable.unwrap_or_default()
341            );
342            tokio::time::sleep(Duration::from_millis(backoff)).await;
343        }
344    }
345}
346
347async fn check_success(resp: reqwest::Response) -> anyhow::Result<reqwest::Response> {
348    let st = resp.status();
349    if st.is_success() {
350        return Ok(resp);
351    }
352    let body = resp.text().await.unwrap_or_default();
353    let msg = serde_json::from_str::<ErrorResponse>(&body)
354        .map(|e| e.error)
355        .unwrap_or(body);
356    anyhow::bail!("memory server error {st}: {msg}")
357}
358
359async fn parse_body<R: DeserializeOwned>(resp: reqwest::Response) -> anyhow::Result<R> {
360    let resp = check_success(resp).await?;
361    Ok(resp.json().await?)
362}
crow_memory_sdk/lib.rs

crow_memory_sdk/
lib.rs
