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    /// Absolute transaction Instant — caps MCP tool dispatch budget.
51    transaction_deadline: std::time::Instant,
52}
53
54impl TransactionMcpHandler {
55    /// Construct for a binding (shares state atom with the route table entry).
56    pub fn new(
57        state: Arc<AtomicU8>,
58        tools: ResolvedToolSet,
59        dispatcher: Arc<TransactionToolDispatcher>,
60        transaction_id: TransactionId,
61        exchange_id: ExchangeId,
62        transaction_deadline: std::time::Instant,
63    ) -> Self {
64        let tool_defs = tool_definitions_from_resolved(&tools);
65        Self {
66            state,
67            tools,
68            dispatcher,
69            transaction_id,
70            exchange_id,
71            tool_defs,
72            transaction_deadline,
73        }
74    }
75
76    fn ensure_active(&self) -> Result<(), McpError> {
77        if self.state.load(Ordering::SeqCst) == STATE_ACTIVE {
78            Ok(())
79        } else {
80            Err(McpError::new(
81                ErrorCode::INVALID_REQUEST,
82                "MCP capability is not active",
83                None,
84            ))
85        }
86    }
87
88    /// Direct list (unit tests without full MCP session).
89    pub fn list_tool_defs(&self) -> Result<Vec<Tool>, McpError> {
90        self.ensure_active()?;
91        Ok(self.tool_defs.clone())
92    }
93
94    /// Direct call through the shared dispatcher.
95    pub async fn call_tool_direct(
96        &self,
97        name: &str,
98        arguments: Option<serde_json::Map<String, serde_json::Value>>,
99    ) -> Result<CallToolResult, McpError> {
100        self.ensure_active()?;
101        let tool_name = ToolName::try_new(name)
102            .map_err(|_| McpError::new(ErrorCode::INVALID_PARAMS, "invalid tool name", None))?;
103        if !self.tools.contains_name(&tool_name) {
104            return Err(McpError::new(
105                ErrorCode::METHOD_NOT_FOUND,
106                "tool not in resolved set",
107                None,
108            ));
109        }
110        let args = arguments.unwrap_or_default();
111        let arguments_json = serde_json::Value::Object(args).to_string();
112        let action = ToolActionId::new(format!("mcp:{}:{}", self.transaction_id.as_uuid(), name));
113        let outcome = self
114            .dispatcher
115            .dispatch(DispatchRequest {
116                exchange_id: self.exchange_id,
117                tool_action_id: action,
118                tool_name,
119                provider_tool_call_id: format!("mcp-{}", uuid::Uuid::new_v4()),
120                request_ordinal: 0,
121                arguments_json,
122                transaction_deadline: self.transaction_deadline,
123            })
124            .await;
125        map_dispatch_to_call_result(outcome)
126    }
127}
128
129impl ServerHandler for TransactionMcpHandler {
130    fn get_info(&self) -> ServerInfo {
131        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
132    }
133
134    async fn list_tools(
135        &self,
136        _request: Option<PaginatedRequestParams>,
137        _context: RequestContext<RoleServer>,
138    ) -> Result<ListToolsResult, McpError> {
139        self.ensure_active()?;
140        // Bounded page: full resolved set is already admission-bounded.
141        Ok(ListToolsResult::with_all_items(self.tool_defs.clone()))
142    }
143
144    async fn call_tool(
145        &self,
146        request: CallToolRequestParams,
147        _context: RequestContext<RoleServer>,
148    ) -> Result<CallToolResponse, McpError> {
149        let name = request.name.as_ref();
150        let args = request.arguments;
151        let result = self.call_tool_direct(name, args).await?;
152        Ok(CallToolResponse::Complete(result))
153    }
154
155    fn get_tool(&self, name: &str) -> Option<Tool> {
156        if self.state.load(Ordering::SeqCst) != STATE_ACTIVE {
157            return None;
158        }
159        self.tool_defs.iter().find(|t| t.name == name).cloned()
160    }
161}
162
163fn map_dispatch_to_call_result(outcome: DispatchOutcome) -> Result<CallToolResult, McpError> {
164    match outcome {
165        DispatchOutcome::Canonical { result, .. } => match result.outcome {
166            CanonicalToolResultOutcome::Succeeded(output) => {
167                let text = match output {
168                    monoloop_contracts::CanonicalToolOutput::Json(v) => {
169                        serde_json::to_string(&v).unwrap_or_else(|_| "{}".into())
170                    }
171                    monoloop_contracts::CanonicalToolOutput::Text(t) => t,
172                };
173                Ok(CallToolResult::success(vec![ContentBlock::text(text)]))
174            }
175            CanonicalToolResultOutcome::DomainFailed(err) => {
176                Ok(CallToolResult::error(vec![ContentBlock::text(format!(
177                    "{}: {}",
178                    err.code, err.message
179                ))]))
180            }
181        },
182        DispatchOutcome::Rejected { code, message, .. } => {
183            // Invalid args: tool-level error visible to the agent.
184            Ok(CallToolResult::error(vec![ContentBlock::text(format!(
185                "{code}: {message}"
186            ))]))
187        }
188        DispatchOutcome::RuntimeFailed { code, .. } => Err(McpError::new(
189            ErrorCode::INTERNAL_ERROR,
190            format!("tool runtime failure: {code}"),
191            None,
192        )),
193    }
194}
monoloop_loop/transaction/mcp/handler.rs

monoloop_loop/transaction/mcp/
handler.rs
