cookie_agent_protocol/session/
server.rs

1use std::{
2    collections::HashSet,
3    sync::{Arc, Mutex},
4};
5
6use async_trait::async_trait;
7use serde::{Deserialize, Serialize};
8use serde_json::{Value, json};
9use tokio::sync::mpsc;
10use tokio_util::sync::CancellationToken;
11
12use crate::{
13    AgentUsageParams, AgentUsageResult, ApprovalListParams, ApprovalListResult,
14    ApprovalRespondParams, ApprovalRespondResult, ClientHello, ClientRenameId, ErrorResponse,
15    EventsSubscribeParams, EventsSubscribeResult, GlobalUsageParams, GlobalUsageResult,
16    JsonRpcError, JsonRpcId, JsonRpcVersion, McpAuthBeginParams, McpAuthBeginResult,
17    McpAuthCancelParams, McpAuthCancelResult, McpServerAddParams, McpServerEditParams,
18    McpServerListParams, McpServerListResult, McpServerMutationResult, McpServerNameParams,
19    McpServerPersistParams, McpServerSetEnabledParams, MessageFrame, Notification,
20    ProviderConnectParams, ProviderConnectResult, ProviderDisconnectParams,
21    ProviderDisconnectResult, Request, Response, RunCancelParams, RunCancelResult,
22    RunRecallSteerParams, RunRecallSteerResult, RunStartParams, RunStartResult, RunSteerParams,
23    RunSteerResult, RunToolStdinParams, RunToolStdinResult, RuntimeSnapshotGetParams,
24    RuntimeSnapshotResult, ServerHello, SessionChildrenParams, SessionChildrenResult,
25    SessionCompactParams, SessionCompactResult, SessionCreateParams, SessionCreateResult,
26    SessionForkParams, SessionForkResult, SessionGetParams, SessionGetResult, SessionGoalGetParams,
27    SessionGoalGetResult, SessionGoalLifecycleParams, SessionGoalLifecycleResult,
28    SessionGoalSetParams, SessionGoalSetResult, SessionId, SessionListParams, SessionListResult,
29    SessionPermissionClearParams, SessionPermissionGetParams, SessionPermissionGetResult,
30    SessionPermissionMutationResult, SessionPermissionSetParams, SessionProducersParams,
31    SessionProducersResult, SessionRenameChange, SessionRenameError, SessionRenameErrorCode,
32    SessionRenameParams, SessionRenameResult, SessionResumeParams, SessionResumeResult,
33    SessionRevertParams, SessionRevertResult, SessionSetPermissionModeParams,
34    SessionSetPermissionModeResult, SessionTitle, SessionTreeParams, SessionTreeResult,
35    SessionTreeUsageResult, SessionUsageParams, SessionUsageResult, SkillsGetParams,
36    SkillsGetResult, SkillsListParams, SkillsListResult, SuccessResponse, Transport,
37    TransportError,
38};
39
40const OUTBOUND_QUEUE_CAPACITY: usize = 512;
41const MAX_RAW_RENAME_PARAMS_BYTES: usize = 4 * 1024;
42
43/// Per-connection facilities available to a server implementation.
44#[derive(Clone)]
45pub struct ServerContext {
46    notifications: mpsc::Sender<Value>,
47    shutdown: CancellationToken,
48    subscribed_sessions: Arc<Mutex<HashSet<SessionId>>>,
49}
50
51impl ServerContext {
52    /// Emit one JSON-RPC notification in connection order.
53    pub async fn notify<T: Serialize>(
54        &self,
55        method: &str,
56        params: &T,
57    ) -> Result<(), TransportError> {
58        let params = serde_json::to_value(params)?;
59        let notification = serde_json::to_value(Notification::new(method, Some(params)))?;
60        tokio::select! {
61            () = self.shutdown.cancelled() => Err(TransportError::Closed),
62            result = self.notifications.send(notification) => result.map_err(|_| TransportError::Closed),
63        }
64    }
65
66    /// A cancellation token scoped to this protocol connection.
67    #[must_use]
68    pub fn shutdown(&self) -> CancellationToken {
69        self.shutdown.clone()
70    }
71
72    /// Marks a session as authorized for connection-scoped live notifications.
73    pub fn register_session_subscription(&self, session_id: SessionId) {
74        self.subscribed_sessions
75            .lock()
76            .unwrap_or_else(|poisoned| poisoned.into_inner())
77            .insert(session_id);
78    }
79
80    /// Returns whether this connection has successfully subscribed to the session.
81    #[must_use]
82    pub fn is_session_subscribed(&self, session_id: SessionId) -> bool {
83        self.subscribed_sessions
84            .lock()
85            .unwrap_or_else(|poisoned| poisoned.into_inner())
86            .contains(&session_id)
87    }
88}
89
90#[cfg(feature = "test-support")]
91#[doc(hidden)]
92#[must_use]
93pub fn test_server_context() -> (ServerContext, mpsc::Receiver<Value>) {
94    let (notifications, receiver) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
95    (
96        ServerContext {
97            notifications,
98            shutdown: CancellationToken::new(),
99            subscribed_sessions: Arc::new(Mutex::new(HashSet::new())),
100        },
101        receiver,
102    )
103}
104
105/// Server end of the protocol contract after session-level validation.
106#[async_trait]
107pub trait ServerProtocol: Send + Sync + 'static {
108    /// Called after the exact-version handshake response has been sent.
109    async fn connected(&self, _context: ServerContext) {}
110
111    async fn create_session(
112        &self,
113        params: SessionCreateParams,
114    ) -> Result<SessionCreateResult, ServerFault>;
115    async fn list_sessions(
116        &self,
117        params: SessionListParams,
118    ) -> Result<SessionListResult, ServerFault>;
119    async fn get_session(&self, params: SessionGetParams) -> Result<SessionGetResult, ServerFault>;
120    /// Returns the root session's current goal, if any.
121    ///
122    /// Implementations must reject child sessions. Goal access is root-only and
123    /// is not a configurable permission surface.
124    async fn get_session_goal(
125        &self,
126        params: SessionGoalGetParams,
127    ) -> Result<SessionGoalGetResult, ServerFault>;
128    /// Activates a user-owned goal on a root session.
129    ///
130    /// Implementations must reject child sessions and must reject activation
131    /// while the current goal is active or paused. This user RPC must not be
132    /// reused as a model/tool mutation path.
133    async fn set_session_goal(
134        &self,
135        params: SessionGoalSetParams,
136    ) -> Result<SessionGoalSetResult, ServerFault>;
137    /// Applies a user-owned lifecycle transition to a root session goal.
138    ///
139    /// Implementations must reject child sessions, terminal goals, mismatched
140    /// goal IDs, and stale `expected_revision` values. This user RPC must not be
141    /// exposed to model tools.
142    async fn change_session_goal_lifecycle(
143        &self,
144        params: SessionGoalLifecycleParams,
145    ) -> Result<SessionGoalLifecycleResult, ServerFault>;
146    /// Returns the runtime-only producer registry for the requested session.
147    async fn session_producers(
148        &self,
149        params: SessionProducersParams,
150    ) -> Result<SessionProducersResult, ServerFault>;
151    async fn session_usage(
152        &self,
153        params: SessionUsageParams,
154    ) -> Result<SessionUsageResult, ServerFault>;
155    async fn session_tree_usage(
156        &self,
157        params: SessionUsageParams,
158    ) -> Result<SessionTreeUsageResult, ServerFault>;
159    async fn agent_usage(&self, params: AgentUsageParams) -> Result<AgentUsageResult, ServerFault>;
160    async fn global_usage(
161        &self,
162        params: GlobalUsageParams,
163    ) -> Result<GlobalUsageResult, ServerFault>;
164    async fn session_children(
165        &self,
166        params: SessionChildrenParams,
167    ) -> Result<SessionChildrenResult, ServerFault>;
168    async fn session_tree(
169        &self,
170        params: SessionTreeParams,
171    ) -> Result<SessionTreeResult, ServerFault>;
172    async fn resume_session(
173        &self,
174        params: SessionResumeParams,
175    ) -> Result<SessionResumeResult, ServerFault>;
176    async fn rename_session(
177        &self,
178        params: SessionRenameParams,
179    ) -> Result<SessionRenameResult, ServerFault>;
180    async fn set_permission_mode(
181        &self,
182        params: SessionSetPermissionModeParams,
183    ) -> Result<SessionSetPermissionModeResult, ServerFault>;
184    async fn get_session_permissions(
185        &self,
186        params: SessionPermissionGetParams,
187    ) -> Result<SessionPermissionGetResult, ServerFault>;
188    async fn set_session_permission(
189        &self,
190        params: SessionPermissionSetParams,
191    ) -> Result<SessionPermissionMutationResult, ServerFault>;
192    async fn clear_session_permission(
193        &self,
194        params: SessionPermissionClearParams,
195    ) -> Result<SessionPermissionMutationResult, ServerFault>;
196    async fn list_skills(&self, params: SkillsListParams) -> Result<SkillsListResult, ServerFault>;
197    async fn get_skill(&self, params: SkillsGetParams) -> Result<SkillsGetResult, ServerFault>;
198    async fn compact_session(
199        &self,
200        params: SessionCompactParams,
201    ) -> Result<SessionCompactResult, ServerFault>;
202    async fn revert_session(
203        &self,
204        params: SessionRevertParams,
205    ) -> Result<SessionRevertResult, ServerFault>;
206    async fn fork_session(
207        &self,
208        params: SessionForkParams,
209    ) -> Result<SessionForkResult, ServerFault>;
210    async fn start_run(&self, params: RunStartParams) -> Result<RunStartResult, ServerFault>;
211    async fn steer_run(&self, params: RunSteerParams) -> Result<RunSteerResult, ServerFault>;
212    async fn recall_steer(
213        &self,
214        params: RunRecallSteerParams,
215    ) -> Result<RunRecallSteerResult, ServerFault>;
216    async fn cancel_run(&self, params: RunCancelParams) -> Result<RunCancelResult, ServerFault>;
217    async fn tool_stdin(
218        &self,
219        params: RunToolStdinParams,
220    ) -> Result<RunToolStdinResult, ServerFault>;
221    async fn subscribe_events(
222        &self,
223        params: EventsSubscribeParams,
224        context: &ServerContext,
225    ) -> Result<EventsSubscribeResult, ServerFault>;
226    async fn respond_approval(
227        &self,
228        params: ApprovalRespondParams,
229    ) -> Result<ApprovalRespondResult, ServerFault>;
230    async fn list_approvals(
231        &self,
232        params: ApprovalListParams,
233    ) -> Result<ApprovalListResult, ServerFault>;
234    async fn begin_mcp_auth(
235        &self,
236        params: McpAuthBeginParams,
237    ) -> Result<McpAuthBeginResult, ServerFault>;
238    async fn cancel_mcp_auth(
239        &self,
240        params: McpAuthCancelParams,
241    ) -> Result<McpAuthCancelResult, ServerFault>;
242    async fn list_mcp_servers(
243        &self,
244        params: McpServerListParams,
245    ) -> Result<McpServerListResult, ServerFault>;
246    async fn add_mcp_server(
247        &self,
248        params: McpServerAddParams,
249    ) -> Result<McpServerMutationResult, ServerFault>;
250    async fn edit_mcp_server(
251        &self,
252        params: McpServerEditParams,
253    ) -> Result<McpServerMutationResult, ServerFault>;
254    async fn remove_mcp_server(
255        &self,
256        params: McpServerNameParams,
257    ) -> Result<McpServerMutationResult, ServerFault>;
258    async fn set_mcp_server_enabled(
259        &self,
260        params: McpServerSetEnabledParams,
261    ) -> Result<McpServerMutationResult, ServerFault>;
262    async fn reconnect_mcp_server(
263        &self,
264        params: McpServerNameParams,
265    ) -> Result<McpServerMutationResult, ServerFault>;
266    async fn persist_mcp_server(
267        &self,
268        params: McpServerPersistParams,
269    ) -> Result<McpServerMutationResult, ServerFault>;
270    async fn runtime_snapshot(
271        &self,
272        params: RuntimeSnapshotGetParams,
273    ) -> Result<RuntimeSnapshotResult, ServerFault>;
274    async fn connect_provider(
275        &self,
276        params: ProviderConnectParams,
277    ) -> Result<ProviderConnectResult, ServerFault>;
278    async fn disconnect_provider(
279        &self,
280        params: ProviderDisconnectParams,
281    ) -> Result<ProviderDisconnectResult, ServerFault>;
282}
283
284/// JSON-RPC failure returned by a [`ServerProtocol`] implementation.
285#[derive(Debug)]
286pub struct ServerFault {
287    pub code: i32,
288    pub message: &'static str,
289    pub data: Option<Value>,
290}
291
292impl ServerFault {
293    #[must_use]
294    pub const fn new(code: i32, message: &'static str, data: Option<Value>) -> Self {
295        Self {
296            code,
297            message,
298            data,
299        }
300    }
301
302    #[must_use]
303    pub const fn method_not_found() -> Self {
304        Self::new(-32601, "method not found", None)
305    }
306
307    #[must_use]
308    pub const fn invalid_params() -> Self {
309        Self::new(-32602, "invalid params", None)
310    }
311
312    #[must_use]
313    pub const fn internal() -> Self {
314        Self::new(-32603, "internal error", None)
315    }
316}
317
318struct ConnectionShutdown(CancellationToken);
319
320impl Drop for ConnectionShutdown {
321    fn drop(&mut self) {
322        self.0.cancel();
323    }
324}
325
326/// Run one complete server-side protocol session over a transport.
327pub async fn serve<T, S>(
328    server: Arc<S>,
329    mut transport: T,
330    shutdown: CancellationToken,
331) -> Result<(), TransportError>
332where
333    T: Transport,
334    S: ServerProtocol,
335{
336    let (notifications, mut notification_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
337    let context = ServerContext {
338        notifications,
339        shutdown,
340        subscribed_sessions: Arc::new(Mutex::new(HashSet::new())),
341    };
342    let _guard = ConnectionShutdown(context.shutdown());
343    let mut handshaken = false;
344    loop {
345        tokio::select! {
346            () = context.shutdown.cancelled() => return Ok(()),
347            incoming = transport.recv() => {
348                let Some(frame) = incoming? else { return Ok(()); };
349                let incoming = match parse_incoming(frame).and_then(classify_incoming) {
350                    Ok(incoming) => incoming,
351                    Err(fault) => {
352                        transport.send(MessageFrame::Value(error_response(None, fault)?)).await?;
353                        continue;
354                    }
355                };
356                match incoming {
357                    Incoming::Request { id, method, params } => {
358                        let (response, connected) = if !handshaken {
359                            if method != "handshake" {
360                                (error_response(Some(id), handshake_required())?, false)
361                            } else {
362                                match decode_handshake(params) {
363                                    Ok(()) => {
364                                        handshaken = true;
365                                        (success_response(id, &ServerHello { protocol_version: crate::ProtocolVersion::current() })?, true)
366                                    }
367                                    Err(fault) => (error_response(Some(id), fault)?, false),
368                                }
369                            }
370                        } else {
371                            let result = if method == "handshake" {
372                                decode_handshake(params).map(|()| {
373                                    serde_json::to_value(ServerHello {
374                                        protocol_version: crate::ProtocolVersion::current(),
375                                    })
376                                    .expect("server hello serializes")
377                                })
378                            } else {
379                                dispatch(server.as_ref(), &method, params, &context, true).await
380                            };
381                            (match result {
382                                Ok(value) => success_response(id, &value)?,
383                                Err(fault) => error_response(Some(id), fault)?,
384                            }, false)
385                        };
386                        transport.send(MessageFrame::Value(response)).await?;
387                        if connected {
388                            server.connected(context.clone()).await;
389                        }
390                    }
391                    Incoming::Notification { method, params } => {
392                        if handshaken {
393                            if method == "handshake" {
394                                let _ = decode_handshake(params);
395                            } else {
396                                let _ = dispatch(server.as_ref(), &method, params, &context, false).await;
397                            }
398                        }
399                    }
400                }
401            }
402            Some(notification) = notification_rx.recv() => {
403                transport.send(MessageFrame::Value(notification)).await?;
404            }
405        }
406    }
407}
408
409async fn dispatch<S: ServerProtocol>(
410    server: &S,
411    method: &str,
412    params: Option<Value>,
413    context: &ServerContext,
414    has_request_id: bool,
415) -> Result<Value, ServerFault> {
416    match method {
417        "session.create" => value(server.create_session(decode(params)?).await?),
418        "session.list" => value(server.list_sessions(decode_default(params)?).await?),
419        "session.get" => value(server.get_session(decode(params)?).await?),
420        crate::SESSION_GOAL_GET_METHOD => value(server.get_session_goal(decode(params)?).await?),
421        crate::SESSION_GOAL_SET_METHOD => value(server.set_session_goal(decode(params)?).await?),
422        crate::SESSION_GOAL_LIFECYCLE_METHOD => value(
423            server
424                .change_session_goal_lifecycle(decode(params)?)
425                .await?,
426        ),
427        crate::SESSION_PRODUCERS_METHOD => value(server.session_producers(decode(params)?).await?),
428        "session.usage" => value(server.session_usage(decode(params)?).await?),
429        "session.tree_usage" => value(server.session_tree_usage(decode(params)?).await?),
430        "agent.usage" => value(server.agent_usage(decode(params)?).await?),
431        "usage.global" => value(server.global_usage(decode_default(params)?).await?),
432        "session.children" => value(server.session_children(decode(params)?).await?),
433        "session.tree" => value(server.session_tree(decode(params)?).await?),
434        "session.resume" => value(server.resume_session(decode(params)?).await?),
435        "session.rename" => value(server.rename_session(decode_rename(params)?).await?),
436        "session.set_permission_mode" => value(server.set_permission_mode(decode(params)?).await?),
437        "session.permission.get" => value(server.get_session_permissions(decode(params)?).await?),
438        "session.permission.set" => value(server.set_session_permission(decode(params)?).await?),
439        "session.permission.clear" => {
440            value(server.clear_session_permission(decode(params)?).await?)
441        }
442        "skills.list" => value(server.list_skills(decode(params)?).await?),
443        "skills.get" => value(server.get_skill(decode(params)?).await?),
444        "session.compact" => value(server.compact_session(decode(params)?).await?),
445        "session.revert" => value(server.revert_session(decode(params)?).await?),
446        "session.fork" => value(server.fork_session(decode(params)?).await?),
447        "run.start" => value(server.start_run(decode(params)?).await?),
448        "run.steer" => value(server.steer_run(decode(params)?).await?),
449        "run.recall_steer" => value(server.recall_steer(decode(params)?).await?),
450        "run.cancel" => value(server.cancel_run(decode(params)?).await?),
451        "run.tool_stdin" => value(server.tool_stdin(decode(params)?).await?),
452        "events.subscribe" => value(server.subscribe_events(decode(params)?, context).await?),
453        "approval.respond" => value(server.respond_approval(decode(params)?).await?),
454        "approval.list" => value(server.list_approvals(decode(params)?).await?),
455        "mcp.auth.begin" => value(server.begin_mcp_auth(decode(params)?).await?),
456        "mcp.auth.cancel" => value(server.cancel_mcp_auth(decode(params)?).await?),
457        "mcp.server.list" => value(server.list_mcp_servers(decode_default(params)?).await?),
458        "mcp.server.add" => value(server.add_mcp_server(decode(params)?).await?),
459        "mcp.server.edit" => value(server.edit_mcp_server(decode(params)?).await?),
460        "mcp.server.remove" => value(server.remove_mcp_server(decode(params)?).await?),
461        "mcp.server.set_enabled" => value(server.set_mcp_server_enabled(decode(params)?).await?),
462        "mcp.server.reconnect" => value(server.reconnect_mcp_server(decode(params)?).await?),
463        "mcp.server.persist" => value(server.persist_mcp_server(decode(params)?).await?),
464        crate::RUNTIME_SNAPSHOT_GET_METHOD if has_request_id => {
465            value(server.runtime_snapshot(decode_default(params)?).await?)
466        }
467        crate::PROVIDER_CONNECT_METHOD if has_request_id => {
468            value(server.connect_provider(decode(params)?).await?)
469        }
470        crate::PROVIDER_DISCONNECT_METHOD if has_request_id => {
471            value(server.disconnect_provider(decode(params)?).await?)
472        }
473        crate::RUNTIME_SNAPSHOT_GET_METHOD
474        | crate::PROVIDER_CONNECT_METHOD
475        | crate::PROVIDER_DISCONNECT_METHOD => {
476            Err(ServerFault::new(-32600, "request id required", None))
477        }
478        _ => Err(ServerFault::method_not_found()),
479    }
480}
481
482fn decode<T: serde::de::DeserializeOwned>(params: Option<Value>) -> Result<T, ServerFault> {
483    serde_json::from_value(params.unwrap_or(Value::Null)).map_err(|_| ServerFault::invalid_params())
484}
485
486fn decode_default<T>(params: Option<Value>) -> Result<T, ServerFault>
487where
488    T: serde::de::DeserializeOwned + Default,
489{
490    match params {
491        Some(params) => serde_json::from_value(params).map_err(|_| ServerFault::invalid_params()),
492        None => Ok(T::default()),
493    }
494}
495
496#[derive(Deserialize)]
497#[serde(deny_unknown_fields)]
498struct RawSessionRenameParams {
499    session_id: SessionId,
500    client_rename_id: ClientRenameId,
501    change: RawSessionRenameChange,
502}
503
504#[derive(Deserialize)]
505#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
506enum RawSessionRenameChange {
507    Set { title: String },
508    Clear,
509    Reset,
510}
511
512fn decode_rename(params: Option<Value>) -> Result<SessionRenameParams, ServerFault> {
513    let params = params.unwrap_or(Value::Null);
514    if let Ok(request) = serde_json::from_value::<SessionRenameParams>(params.clone()) {
515        return Ok(request);
516    }
517    if serde_json::to_vec(&params)
518        .map_err(|_| ServerFault::invalid_params())?
519        .len()
520        > MAX_RAW_RENAME_PARAMS_BYTES
521    {
522        return Err(ServerFault::invalid_params());
523    }
524    let raw: RawSessionRenameParams =
525        serde_json::from_value(params).map_err(|_| ServerFault::invalid_params())?;
526    let invalid_title = rename_invalid_title(&raw);
527    let change = match raw.change {
528        RawSessionRenameChange::Set { title } => SessionRenameChange::Set {
529            title: SessionTitle::new(title).map_err(|_| invalid_title)?,
530        },
531        RawSessionRenameChange::Clear => SessionRenameChange::Clear,
532        RawSessionRenameChange::Reset => SessionRenameChange::Reset,
533    };
534    Ok(SessionRenameParams {
535        session_id: raw.session_id,
536        client_rename_id: raw.client_rename_id,
537        change,
538    })
539}
540
541fn rename_invalid_title(raw: &RawSessionRenameParams) -> ServerFault {
542    ServerFault::new(
543        -32602,
544        "session rename rejected",
545        serde_json::to_value(SessionRenameError {
546            code: SessionRenameErrorCode::InvalidTitle,
547            session_id: raw.session_id,
548            client_rename_id: raw.client_rename_id.clone(),
549        })
550        .ok(),
551    )
552}
553
554fn value<T: Serialize>(result: T) -> Result<Value, ServerFault> {
555    serde_json::to_value(result).map_err(|_| ServerFault::internal())
556}
557
558enum Incoming {
559    Request {
560        id: JsonRpcId,
561        method: String,
562        params: Option<Value>,
563    },
564    Notification {
565        method: String,
566        params: Option<Value>,
567    },
568}
569
570fn parse_incoming(frame: MessageFrame) -> Result<Value, ServerFault> {
571    match frame {
572        MessageFrame::Value(value) => Ok(value),
573        MessageFrame::Text(text) => {
574            serde_json::from_str(&text).map_err(|_| ServerFault::new(-32700, "parse error", None))
575        }
576    }
577}
578
579fn classify_incoming(value: Value) -> Result<Incoming, ServerFault> {
580    let object = value.as_object().ok_or_else(|| {
581        ServerFault::new(
582            -32600,
583            "invalid request",
584            Some(json!({ "detail": "a batch or scalar is not a request" })),
585        )
586    })?;
587    if object.contains_key("id") {
588        let request: Request = serde_json::from_value(value).map_err(|_| {
589            ServerFault::new(
590                -32600,
591                "invalid request",
592                Some(json!({ "detail": "invalid JSON-RPC request envelope" })),
593            )
594        })?;
595        if request.method.is_empty() {
596            return Err(ServerFault::new(
597                -32600,
598                "invalid request",
599                Some(json!({ "detail": "method must be non-empty" })),
600            ));
601        }
602        Ok(Incoming::Request {
603            id: request.id,
604            method: request.method,
605            params: request.params,
606        })
607    } else {
608        let notification: Notification = serde_json::from_value(value).map_err(|_| {
609            ServerFault::new(
610                -32600,
611                "invalid request",
612                Some(json!({ "detail": "invalid JSON-RPC notification envelope" })),
613            )
614        })?;
615        if notification.method.is_empty() {
616            return Err(ServerFault::new(
617                -32600,
618                "invalid request",
619                Some(json!({ "detail": "method must be non-empty" })),
620            ));
621        }
622        Ok(Incoming::Notification {
623            method: notification.method,
624            params: notification.params,
625        })
626    }
627}
628
629fn decode_handshake(params: Option<Value>) -> Result<(), ServerFault> {
630    serde_json::from_value::<ClientHello>(params.unwrap_or(Value::Null))
631        .map(|_| ())
632        .map_err(|_| ServerFault::invalid_params())
633}
634
635fn handshake_required() -> ServerFault {
636    ServerFault::new(-32001, "handshake required", None)
637}
638
639fn success_response<T: Serialize>(id: JsonRpcId, result: &T) -> Result<Value, TransportError> {
640    Ok(serde_json::to_value(Response::Success(SuccessResponse {
641        jsonrpc: JsonRpcVersion::current(),
642        id,
643        result: serde_json::to_value(result)?,
644    }))?)
645}
646
647fn error_response(id: Option<JsonRpcId>, fault: ServerFault) -> Result<Value, TransportError> {
648    match id {
649        Some(id) => Ok(serde_json::to_value(Response::Error(ErrorResponse {
650            jsonrpc: JsonRpcVersion::current(),
651            id,
652            error: JsonRpcError {
653                code: fault.code,
654                message: fault.message.into(),
655                data: fault.data,
656            },
657        }))?),
658        None => Ok(json!({
659            "jsonrpc": "2.0",
660            "id": null,
661            "error": { "code": fault.code, "message": fault.message, "data": fault.data },
662        })),
663    }
664}
cookie_agent_protocol/session/server.rs

cookie_agent_protocol/session/
server.rs
