mcpr-integrations 0.4.70

External integrations for mcpr: cloud event sink, API client, and SQLite request storage
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

//! SQLite event sink — writes proxy events to the local SQLite store.
//!
//! Converts `ProxyEvent` variants into store operations:
//! - `Request` → insert into `requests` table + update session counters
//! - `SessionStart` → insert into `sessions` table
//! - `SessionEnd` → update `ended_at` on the session
//! - `Heartbeat` → ignored (not stored locally)

use mcpr_core::event::{EventSink, ProxyEvent, SchemaVersionCreatedEvent};

use crate::store::engine::Store;
use crate::store::event::{
    RequestEvent as StoreRequestEvent, RequestStatus, SchemaVersionEvent,
    SessionEvent as StoreSessionEvent, StoreEvent,
};

/// Event sink that writes to the SQLite store.
///
/// Wraps the existing `Store` and converts `ProxyEvent` → `StoreEvent`.
pub struct SqliteSink {
    store: Store,
}

impl SqliteSink {
    pub fn new(store: Store) -> Self {
        Self { store }
    }

    /// Shutdown the underlying store (flush pending writes).
    pub fn shutdown(&mut self) {
        self.store.shutdown();
    }
}

impl EventSink for SqliteSink {
    fn on_event(&self, event: &ProxyEvent) {
        match event {
            ProxyEvent::Request(e) => {
                let status = if e.error_code.is_some() || e.status >= 500 {
                    RequestStatus::Error
                } else {
                    RequestStatus::Ok
                };

                self.store.record(StoreEvent::Request(StoreRequestEvent {
                    request_id: e.id.clone(),
                    ts: e.ts,
                    proxy: e.proxy.clone(),
                    session_id: e.session_id.clone(),
                    method: e.mcp_method.clone().unwrap_or_else(|| e.method.clone()),
                    tool: e.tool.clone(),
                    resource_uri: e.resource_uri.clone(),
                    prompt_name: e.prompt_name.clone(),
                    latency_us: e.latency_us as i64,
                    status,
                    error_code: e.error_code.clone(),
                    error_msg: e.error_msg.clone(),
                    bytes_in: e.request_size.map(|s| s as i64),
                    bytes_out: e.response_size.map(|s| s as i64),
                }));
            }
            ProxyEvent::SessionStart(e) => {
                self.store.record(StoreEvent::Session(StoreSessionEvent {
                    session_id: e.session_id.clone(),
                    proxy: e.proxy.clone(),
                    started_at: e.ts,
                    client_name: e.client_name.clone(),
                    client_version: e.client_version.clone(),
                    client_platform: e.client_platform.clone(),
                }));
            }
            ProxyEvent::SessionEnd(e) => {
                self.store.record(StoreEvent::SessionClosed {
                    session_id: e.session_id.clone(),
                    ended_at: e.ts,
                });
            }
            ProxyEvent::Heartbeat(_) => {
                // Heartbeats are not stored locally.
            }
            ProxyEvent::SchemaVersionCreated(e) => {
                self.store.record(map_schema_version(e));
            }
        }
    }

    fn name(&self) -> &'static str {
        "sqlite"
    }
}

fn map_schema_version(e: &SchemaVersionCreatedEvent) -> StoreEvent {
    StoreEvent::SchemaVersion(SchemaVersionEvent {
        ts: e.ts,
        proxy: e.upstream_id.clone(),
        upstream_url: e.upstream_url.clone(),
        method: e.method.clone(),
        payload: e.payload.to_string(),
        content_hash: e.content_hash.clone(),
    })
}

#[cfg(test)]
#[allow(non_snake_case)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn map_schema_version__copies_fields_correctly() {
        let event = SchemaVersionCreatedEvent {
            ts: 1_700_000_000_000,
            upstream_id: "api".into(),
            upstream_url: "http://localhost:9000".into(),
            method: "tools/list".into(),
            version: 3,
            version_id: "abc123def4567890".into(),
            content_hash: "abc123def4567890cafebabe".into(),
            payload: json!({"tools": [{"name": "search"}]}),
        };

        let StoreEvent::SchemaVersion(sv) = map_schema_version(&event) else {
            panic!("expected StoreEvent::SchemaVersion");
        };
        assert_eq!(sv.ts, 1_700_000_000_000);
        assert_eq!(sv.proxy, "api");
        assert_eq!(sv.upstream_url, "http://localhost:9000");
        assert_eq!(sv.method, "tools/list");
        assert_eq!(sv.content_hash, "abc123def4567890cafebabe");
        assert!(sv.payload.contains("search"));
    }

    #[test]
    fn map_schema_version__upstream_id_maps_to_proxy_column() {
        let event = SchemaVersionCreatedEvent {
            ts: 0,
            upstream_id: "proxy-alpha".into(),
            upstream_url: "".into(),
            method: "initialize".into(),
            version: 1,
            version_id: "0".into(),
            content_hash: "0".into(),
            payload: json!({}),
        };
        let StoreEvent::SchemaVersion(sv) = map_schema_version(&event) else {
            panic!();
        };
        assert_eq!(sv.proxy, "proxy-alpha");
    }
}