monoloop_loop/transaction/mcp/
handler.rs

1//! Per-transaction MCP ServerHandler delegating to TransactionToolDispatcher.
2
3use crate::transaction::dispatcher::{DispatchOutcome, DispatchRequest, TransactionToolDispatcher};
4use crate::transaction::resolved_tools::ResolvedToolSet;
5use monoloop_contracts::{
6    CanonicalToolResultOutcome, ExchangeId, ToolActionId, ToolName, TransactionId,
7};
8use rmcp::{
9    model::{
10        CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorCode,
11        ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool,
12    },
13    service::RequestContext,
14    ErrorData as McpError, RoleServer, ServerHandler,
15};
16use std::sync::atomic::{AtomicU8, Ordering};
17use std::sync::Arc;
18
19const STATE_ACTIVE: u8 = 1;
20
21/// Project resolved tool specs into MCP `tools/list` definitions (parity with encoders).
22pub fn tool_definitions_from_resolved(tools: &ResolvedToolSet) -> Vec<Tool> {
23    tools
24        .specs()
25        .into_iter()
26        .map(|spec| {
27            let schema = spec.input_schema.as_value().clone();
28            let obj = match schema {
29                serde_json::Value::Object(m) => Arc::new(m),
30                _ => Arc::new(serde_json::Map::new()),
31            };
32            Tool::new(
33                spec.name.as_str().to_string(),
34                spec.description.clone(),
35                obj,
36            )
37        })
38        .collect()
39}
40
41/// MCP handler bound to one transaction capability.
42#[derive(Clone)]
43pub struct TransactionMcpHandler {
44    state: Arc<AtomicU8>,
45    tools: ResolvedToolSet,
46    dispatcher: Arc<TransactionToolDispatcher>,
47    transaction_id: TransactionId,
48    exchange_id: ExchangeId,
49    tool_defs: Vec<Tool>,
50}
51
52impl TransactionMcpHandler {
53    /// Construct for a binding (shares state atom with the route table entry).
54    pub fn new(
55        state: Arc<AtomicU8>,
56        tools: ResolvedToolSet,
57        dispatcher: Arc<TransactionToolDispatcher>,
58        transaction_id: TransactionId,
59        exchange_id: ExchangeId,
60    ) -> Self {
61        let tool_defs = tool_definitions_from_resolved(&tools);
62        Self {
63            state,
64            tools,
65            dispatcher,
66            transaction_id,
67            exchange_id,
68            tool_defs,
69        }
70    }
71
72    fn ensure_active(&self) -> Result<(), McpError> {
73        if self.state.load(Ordering::SeqCst) == STATE_ACTIVE {
74            Ok(())
75        } else {
76            Err(McpError::new(
77                ErrorCode::INVALID_REQUEST,
78                "MCP capability is not active",
79                None,
80            ))
81        }
82    }
83
84    /// Direct list (unit tests without full MCP session).
85    pub fn list_tool_defs(&self) -> Result<Vec<Tool>, McpError> {
86        self.ensure_active()?;
87        Ok(self.tool_defs.clone())
88    }
89
90    /// Direct call through the shared dispatcher.
91    pub async fn call_tool_direct(
92        &self,
93        name: &str,
94        arguments: Option<serde_json::Map<String, serde_json::Value>>,
95    ) -> Result<CallToolResult, McpError> {
96        self.ensure_active()?;
97        let tool_name = ToolName::try_new(name)
98            .map_err(|_| McpError::new(ErrorCode::INVALID_PARAMS, "invalid tool name", None))?;
99        if !self.tools.contains_name(&tool_name) {
100            return Err(McpError::new(
101                ErrorCode::METHOD_NOT_FOUND,
102                "tool not in resolved set",
103                None,
104            ));
105        }
106        let args = arguments.unwrap_or_default();
107        let arguments_json = serde_json::Value::Object(args).to_string();
108        let action = ToolActionId::new(format!("mcp:{}:{}", self.transaction_id.as_uuid(), name));
109        let outcome = self
110            .dispatcher
111            .dispatch(DispatchRequest {
112                exchange_id: self.exchange_id,
113                tool_action_id: action,
114                tool_name,
115                provider_tool_call_id: format!("mcp-{}", uuid::Uuid::new_v4()),
116                request_ordinal: 0,
117                arguments_json,
118            })
119            .await;
120        map_dispatch_to_call_result(outcome)
121    }
122}
123
124impl ServerHandler for TransactionMcpHandler {
125    fn get_info(&self) -> ServerInfo {
126        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
127    }
128
129    async fn list_tools(
130        &self,
131        _request: Option<PaginatedRequestParams>,
132        _context: RequestContext<RoleServer>,
133    ) -> Result<ListToolsResult, McpError> {
134        self.ensure_active()?;
135        // Bounded page: full resolved set is already admission-bounded.
136        Ok(ListToolsResult::with_all_items(self.tool_defs.clone()))
137    }
138
139    async fn call_tool(
140        &self,
141        request: CallToolRequestParams,
142        _context: RequestContext<RoleServer>,
143    ) -> Result<CallToolResponse, McpError> {
144        let name = request.name.as_ref();
145        let args = request.arguments;
146        let result = self.call_tool_direct(name, args).await?;
147        Ok(CallToolResponse::Complete(result))
148    }
149
150    fn get_tool(&self, name: &str) -> Option<Tool> {
151        if self.state.load(Ordering::SeqCst) != STATE_ACTIVE {
152            return None;
153        }
154        self.tool_defs.iter().find(|t| t.name == name).cloned()
155    }
156}
157
158fn map_dispatch_to_call_result(outcome: DispatchOutcome) -> Result<CallToolResult, McpError> {
159    match outcome {
160        DispatchOutcome::Canonical { result, .. } => match result.outcome {
161            CanonicalToolResultOutcome::Succeeded(output) => {
162                let text = match output {
163                    monoloop_contracts::CanonicalToolOutput::Json(v) => {
164                        serde_json::to_string(&v).unwrap_or_else(|_| "{}".into())
165                    }
166                    monoloop_contracts::CanonicalToolOutput::Text(t) => t,
167                };
168                Ok(CallToolResult::success(vec![ContentBlock::text(text)]))
169            }
170            CanonicalToolResultOutcome::DomainFailed(err) => {
171                Ok(CallToolResult::error(vec![ContentBlock::text(format!(
172                    "{}: {}",
173                    err.code, err.message
174                ))]))
175            }
176        },
177        DispatchOutcome::Rejected { code, message, .. } => {
178            // Invalid args: tool-level error visible to the agent.
179            Ok(CallToolResult::error(vec![ContentBlock::text(format!(
180                "{code}: {message}"
181            ))]))
182        }
183        DispatchOutcome::RuntimeFailed { code, .. } => Err(McpError::new(
184            ErrorCode::INTERNAL_ERROR,
185            format!("tool runtime failure: {code}"),
186            None,
187        )),
188    }
189}
monoloop_loop/transaction/mcp/handler.rs

monoloop_loop/transaction/mcp/
handler.rs
