Vakyartha documentation

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use serde_json::Value;
5
6use vak_tools::{Tool, ToolContext, ToolOutput};
7
8use crate::manager::McpManager;
9
10type InvocationRecorder = Arc<dyn Fn(&str, &str, bool, u64) + Send + Sync>;
11type CatalogObserver = Arc<dyn Fn(&[(String, Vec<crate::McpToolInfo>)]) + Send + Sync>;
12
13/// One meta-tool exposing every configured MCP server without dumping tool
14/// descriptions into context. The model lists on demand, then calls.
15pub struct McpTool {
16 manager: Arc<McpManager>,
17 allow: Option<Vec<String>>,
18 deny: Vec<String>,
19 invocation_recorder: Option<InvocationRecorder>,
20 catalog_observer: Option<CatalogObserver>,
21}
22
23impl McpTool {
24 /// The broker to every external integration. Claiming `live-data` is
25 /// what lets a live-data reading reach a configured search server;
26 /// servers refine it by declaring their own `serves`.
27 pub const SERVES: &'static [&'static str] = &["live-data", "web", "documents", "messaging"];
28
29 pub fn new(manager: Arc<McpManager>) -> Self {
30 McpTool {
31 manager,
32 allow: None,
33 deny: Vec::new(),
34 invocation_recorder: None,
35 catalog_observer: None,
36 }
37 }
38
39 pub fn with_policy(
40 manager: Arc<McpManager>,
41 allow: Option<Vec<String>>,
42 deny: Vec<String>,
43 ) -> Self {
44 McpTool {
45 manager,
46 allow,
47 deny,
48 invocation_recorder: None,
49 catalog_observer: None,
50 }
51 }
52
53 pub fn with_policy_and_recorder(
54 manager: Arc<McpManager>,
55 allow: Option<Vec<String>>,
56 deny: Vec<String>,
57 recorder: InvocationRecorder,
58 ) -> Self {
59 Self {
60 manager,
61 allow,
62 deny,
63 invocation_recorder: Some(recorder),
64 catalog_observer: None,
65 }
66 }
67
68 pub fn with_catalog_observer(mut self, observer: CatalogObserver) -> Self {
69 self.catalog_observer = Some(observer);
70 self
71 }
72
73 fn matches(patterns: &[String], value: &str) -> bool {
74 patterns.iter().any(|pattern| {
75 globset::Glob::new(pattern)
76 .ok()
77 .is_some_and(|glob| glob.compile_matcher().is_match(value))
78 })
79 }
80
81 fn allowed(&self, server: &str, tool: &str) -> bool {
82 let value = format!("{server}/{tool}");
83 !Self::matches(&self.deny, &value)
84 && self
85 .allow
86 .as_ref()
87 .is_none_or(|allow| Self::matches(allow, &value))
88 }
89}
90
91#[async_trait]
92impl Tool for McpTool {
93 fn name(&self) -> &str {
94 "mcp"
95 }
96
97 fn serves(&self) -> &'static [&'static str] {
98 Self::SERVES
99 }
100
101 fn always_loaded(&self) -> bool {
102 true
103 }
104
105 fn description(&self) -> &str {
106 "Discover and call tools exposed by configured MCP servers. For a current fact or an unknown source URL, check configured servers for a search tool before fetching a page by URL. Always use this broker (never call an MCP tool name directly). Use action \"list\" with a server to get its exact tool names and inputSchemas, then action \"call\" with server, tool, and arguments matching that schema exactly."
107 }
108
109 fn schema(&self) -> Value {
110 // OpenAI function declarations require a top-level object without
111 // oneOf/anyOf. Keep the two actions explicit in descriptions, and
112 // enforce call-only fields again in execute before touching a server.
113 // The configured server names are declaration data, not a catalog
114 // probe, so exposing them here preserves lazy startup while making a
115 // `find_tools` result actionable for models that missed the separate
116 // prompt inventory.
117 let servers = self.reachable_servers();
118 let server_schema = if servers.is_empty() {
119 serde_json::json!({
120 "type": "string",
121 "description": "The configured server to list or call. Required for call."
122 })
123 } else {
124 serde_json::json!({
125 "type": "string",
126 "enum": servers,
127 "description": "A configured, policy-reachable server to list or call. Required for call."
128 })
129 };
130 serde_json::json!({
131 "type": "object",
132 "properties": {
133 "action": {"type": "string", "enum": ["list", "call"], "description": "list: a server's exact tool names and schemas (without a server, just the server names); call: invoke one tool."},
134 "server": server_schema,
135 "tool": {"type": "string", "description": "Required for call: a tool name returned by list for that server."},
136 "arguments": {"type": "object", "description": "For call: arguments matching the discovered tool's inputSchema."}
137 },
138 "required": ["action"],
139 "additionalProperties": false
140 })
141 }
142
143 async fn execute(&self, args: &Value, _ctx: &ToolContext) -> ToolOutput {
144 match args.get("action").and_then(|a| a.as_str()) {
145 Some("list") => self.list(args).await,
146 Some("call") => self.call(args).await,
147 Some(other) => ToolOutput::error(format!("unknown mcp action '{other}'")),
148 None => ToolOutput::error("missing required parameter: action"),
149 }
150 }
151}
152
153impl McpTool {
154 /// Servers this turn may reach: configured, and not wholly denied.
155 fn reachable_servers(&self) -> Vec<String> {
156 self.manager
157 .server_names()
158 .into_iter()
159 .filter(|server| {
160 let may_be_allowed = self.allow.as_ref().is_none_or(|allow| {
161 allow.iter().any(|pattern| {
162 let server_pattern = pattern.split('/').next().unwrap_or_default();
163 globset::Glob::new(server_pattern)
164 .ok()
165 .is_some_and(|glob| glob.compile_matcher().is_match(server))
166 })
167 });
168 may_be_allowed && !Self::matches(&self.deny, &format!("{server}/*"))
169 })
170 .collect()
171 }
172
173 /// `list` without a server connects to nothing: it answers from what the
174 /// pool already knows. With a server it is demand for exactly that one.
175 async fn list(&self, args: &Value) -> ToolOutput {
176 let servers = self.reachable_servers();
177 if servers.is_empty() {
178 return ToolOutput::ok("no MCP servers configured");
179 }
180 let Some(server) = args.get("server").and_then(|s| s.as_str()) else {
181 let observed = self.manager.observations();
182 let mut out =
183 String::from("MCP servers (call list with a server to get its tool schemas):\n");
184 for server in &servers {
185 let known = observed.get(server);
186 match known.and_then(|o| o.tools.as_ref()) {
187 Some(tools) => {
188 let names: Vec<&str> = tools
189 .iter()
190 .filter(|t| self.allowed(server, &t.name))
191 .map(|t| t.name.as_str())
192 .collect();
193 out.push_str(&format!("- {server}: {}\n", names.join(", ")));
194 }
195 None => out.push_str(&format!("- {server}\n")),
196 }
197 if let Some(failure) = known.and_then(|o| o.failure.as_deref()) {
198 out.push_str(&format!(" last attempt failed: {failure}\n"));
199 }
200 }
201 return ToolOutput::ok(out);
202 };
203 if !servers.iter().any(|name| name == server) {
204 return ToolOutput::error(format!(
205 "unknown mcp server '{server}'; available: {}",
206 servers.join(", ")
207 ));
208 }
209 let tools = match self.manager.list_tools(server).await {
210 Ok(tools) => tools,
211 Err(reason) => return ToolOutput::error(format!("mcp list failed: {reason}")),
212 };
213 let visible: Vec<_> = tools
214 .into_iter()
215 .filter(|t| self.allowed(server, &t.name))
216 .collect();
217 if let Some(observer) = &self.catalog_observer {
218 observer(&[(server.to_string(), visible.clone())]);
219 }
220 if visible.is_empty() {
221 return ToolOutput::ok(format!("{server}: (no tools)"));
222 }
223 let mut out = format!("{server}:\n");
224 for t in &visible {
225 let schema =
226 serde_json::to_string(&t.input_schema).unwrap_or_else(|_| "{}".to_string());
227 out.push_str(&format!(
228 " {} — {}\n inputSchema: {schema}\n",
229 t.name, t.description
230 ));
231 }
232 ToolOutput::ok(out)
233 }
234
235 async fn call(&self, args: &Value) -> ToolOutput {
236 let Some(server) = args.get("server").and_then(|s| s.as_str()) else {
237 return ToolOutput::error("missing required parameter: server");
238 };
239 let Some(tool) = args.get("tool").and_then(|t| t.as_str()) else {
240 return ToolOutput::error("missing required parameter: tool");
241 };
242 let arguments = args
243 .get("arguments")
244 .cloned()
245 .unwrap_or(Value::Object(Default::default()));
246
247 if !self.allowed(server, tool) {
248 if let Some(record) = &self.invocation_recorder {
249 record(server, tool, false, 0);
250 }
251 return ToolOutput::error(format!(
252 "MCP capability denied by channel policy: {server}/{tool}"
253 ));
254 }
255
256 let started = std::time::Instant::now();
257 match self.manager.call_tool(server, tool, arguments).await {
258 Ok(text) => {
259 if let Some(record) = &self.invocation_recorder {
260 record(server, tool, true, started.elapsed().as_millis() as u64);
261 }
262 if text.is_empty() {
263 ToolOutput::ok("(empty result)")
264 } else {
265 ToolOutput::ok(self.manager.redact(text))
266 }
267 }
268 Err(e) => {
269 if let Some(record) = &self.invocation_recorder {
270 record(server, tool, false, started.elapsed().as_millis() as u64);
271 }
272 ToolOutput::error(format!(
273 "mcp call failed: {}",
274 self.manager.redact(e.to_string())
275 ))
276 }
277 }
278 }
279}
280

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use serde_json::Value;
5
6use vak_tools::{Tool, ToolContext, ToolOutput};
7
8use crate::manager::McpManager;
9
10type InvocationRecorder = Arc<dyn Fn(&str, &str, bool, u64) + Send + Sync>;
11type CatalogObserver = Arc<dyn Fn(&[(String, Vec<crate::McpToolInfo>)]) + Send + Sync>;
12
13/// One meta-tool exposing every configured MCP server without dumping tool
14/// descriptions into context. The model lists on demand, then calls.
15pub struct McpTool {
16 manager: Arc<McpManager>,
17 allow: Option<Vec<String>>,
18 deny: Vec<String>,
19 invocation_recorder: Option<InvocationRecorder>,
20 catalog_observer: Option<CatalogObserver>,
21}
22
23impl McpTool {
24 /// The broker to every external integration. Claiming `live-data` is
25 /// what lets a live-data reading reach a configured search server;
26 /// servers refine it by declaring their own `serves`.
27 pub const SERVES: &'static [&'static str] = &["live-data", "web", "documents", "messaging"];
28
29 pub fn new(manager: Arc<McpManager>) -> Self {
30 McpTool {
31 manager,
32 allow: None,
33 deny: Vec::new(),
34 invocation_recorder: None,
35 catalog_observer: None,
36 }
37 }
38
39 pub fn with_policy(
40 manager: Arc<McpManager>,
41 allow: Option<Vec<String>>,
42 deny: Vec<String>,
43 ) -> Self {
44 McpTool {
45 manager,
46 allow,
47 deny,
48 invocation_recorder: None,
49 catalog_observer: None,
50 }
51 }
52
53 pub fn with_policy_and_recorder(
54 manager: Arc<McpManager>,
55 allow: Option<Vec<String>>,
56 deny: Vec<String>,
57 recorder: InvocationRecorder,
58 ) -> Self {
59 Self {
60 manager,
61 allow,
62 deny,
63 invocation_recorder: Some(recorder),
64 catalog_observer: None,
65 }
66 }
67
68 pub fn with_catalog_observer(mut self, observer: CatalogObserver) -> Self {
69 self.catalog_observer = Some(observer);
70 self
71 }
72
73 fn matches(patterns: &[String], value: &str) -> bool {
74 patterns.iter().any(|pattern| {
75 globset::Glob::new(pattern)
76 .ok()
77 .is_some_and(|glob| glob.compile_matcher().is_match(value))
78 })
79 }
80
81 fn allowed(&self, server: &str, tool: &str) -> bool {
82 let value = format!("{server}/{tool}");
83 !Self::matches(&self.deny, &value)
84 && self
85 .allow
86 .as_ref()
87 .is_none_or(|allow| Self::matches(allow, &value))
88 }
89}
90
91#[async_trait]
92impl Tool for McpTool {
93 fn name(&self) -> &str {
94 "mcp"
95 }
96
97 fn serves(&self) -> &'static [&'static str] {
98 Self::SERVES
99 }
100
101 fn always_loaded(&self) -> bool {
102 true
103 }
104
105 fn description(&self) -> &str {
106 "Discover and call tools exposed by configured MCP servers. For a current fact or an unknown source URL, check configured servers for a search tool before fetching a page by URL. Always use this broker (never call an MCP tool name directly). Use action \"list\" with a server to get its exact tool names and inputSchemas, then action \"call\" with server, tool, and arguments matching that schema exactly."
107 }
108
109 fn schema(&self) -> Value {
110 // OpenAI function declarations require a top-level object without
111 // oneOf/anyOf. Keep the two actions explicit in descriptions, and
112 // enforce call-only fields again in execute before touching a server.
113 // The configured server names are declaration data, not a catalog
114 // probe, so exposing them here preserves lazy startup while making a
115 // `find_tools` result actionable for models that missed the separate
116 // prompt inventory.
117 let servers = self.reachable_servers();
118 let server_schema = if servers.is_empty() {
119 serde_json::json!({
120 "type": "string",
121 "description": "The configured server to list or call. Required for call."
122 })
123 } else {
124 serde_json::json!({
125 "type": "string",
126 "enum": servers,
127 "description": "A configured, policy-reachable server to list or call. Required for call."
128 })
129 };
130 serde_json::json!({
131 "type": "object",
132 "properties": {
133 "action": {"type": "string", "enum": ["list", "call"], "description": "list: a server's exact tool names and schemas (without a server, just the server names); call: invoke one tool."},
134 "server": server_schema,
135 "tool": {"type": "string", "description": "Required for call: a tool name returned by list for that server."},
136 "arguments": {"type": "object", "description": "For call: arguments matching the discovered tool's inputSchema."}
137 },
138 "required": ["action"],
139 "additionalProperties": false
140 })
141 }
142
143 async fn execute(&self, args: &Value, _ctx: &ToolContext) -> ToolOutput {
144 match args.get("action").and_then(|a| a.as_str()) {
145 Some("list") => self.list(args).await,
146 Some("call") => self.call(args).await,
147 Some(other) => ToolOutput::error(format!("unknown mcp action '{other}'")),
148 None => ToolOutput::error("missing required parameter: action"),
149 }
150 }
151}
152
153impl McpTool {
154 /// Servers this turn may reach: configured, and not wholly denied.
155 fn reachable_servers(&self) -> Vec<String> {
156 self.manager
157 .server_names()
158 .into_iter()
159 .filter(|server| {
160 let may_be_allowed = self.allow.as_ref().is_none_or(|allow| {
161 allow.iter().any(|pattern| {
162 let server_pattern = pattern.split('/').next().unwrap_or_default();
163 globset::Glob::new(server_pattern)
164 .ok()
165 .is_some_and(|glob| glob.compile_matcher().is_match(server))
166 })
167 });
168 may_be_allowed && !Self::matches(&self.deny, &format!("{server}/*"))
169 })
170 .collect()
171 }
172
173 /// `list` without a server connects to nothing: it answers from what the
174 /// pool already knows. With a server it is demand for exactly that one.
175 async fn list(&self, args: &Value) -> ToolOutput {
176 let servers = self.reachable_servers();
177 if servers.is_empty() {
178 return ToolOutput::ok("no MCP servers configured");
179 }
180 let Some(server) = args.get("server").and_then(|s| s.as_str()) else {
181 let observed = self.manager.observations();
182 let mut out =
183 String::from("MCP servers (call list with a server to get its tool schemas):\n");
184 for server in &servers {
185 let known = observed.get(server);
186 match known.and_then(|o| o.tools.as_ref()) {
187 Some(tools) => {
188 let names: Vec<&str> = tools
189 .iter()
190 .filter(|t| self.allowed(server, &t.name))
191 .map(|t| t.name.as_str())
192 .collect();
193 out.push_str(&format!("- {server}: {}\n", names.join(", ")));
194 }
195 None => out.push_str(&format!("- {server}\n")),
196 }
197 if let Some(failure) = known.and_then(|o| o.failure.as_deref()) {
198 out.push_str(&format!(" last attempt failed: {failure}\n"));
199 }
200 }
201 return ToolOutput::ok(out);
202 };
203 if !servers.iter().any(|name| name == server) {
204 return ToolOutput::error(format!(
205 "unknown mcp server '{server}'; available: {}",
206 servers.join(", ")
207 ));
208 }
209 let tools = match self.manager.list_tools(server).await {
210 Ok(tools) => tools,
211 Err(reason) => return ToolOutput::error(format!("mcp list failed: {reason}")),
212 };
213 let visible: Vec<_> = tools
214 .into_iter()
215 .filter(|t| self.allowed(server, &t.name))
216 .collect();
217 if let Some(observer) = &self.catalog_observer {
218 observer(&[(server.to_string(), visible.clone())]);
219 }
220 if visible.is_empty() {
221 return ToolOutput::ok(format!("{server}: (no tools)"));
222 }
223 let mut out = format!("{server}:\n");
224 for t in &visible {
225 let schema =
226 serde_json::to_string(&t.input_schema).unwrap_or_else(|_| "{}".to_string());
227 out.push_str(&format!(
228 " {} — {}\n inputSchema: {schema}\n",
229 t.name, t.description
230 ));
231 }
232 ToolOutput::ok(out)
233 }
234
235 async fn call(&self, args: &Value) -> ToolOutput {
236 let Some(server) = args.get("server").and_then(|s| s.as_str()) else {
237 return ToolOutput::error("missing required parameter: server");
238 };
239 let Some(tool) = args.get("tool").and_then(|t| t.as_str()) else {
240 return ToolOutput::error("missing required parameter: tool");
241 };
242 let arguments = args
243 .get("arguments")
244 .cloned()
245 .unwrap_or(Value::Object(Default::default()));
246
247 if !self.allowed(server, tool) {
248 if let Some(record) = &self.invocation_recorder {
249 record(server, tool, false, 0);
250 }
251 return ToolOutput::error(format!(
252 "MCP capability denied by channel policy: {server}/{tool}"
253 ));
254 }
255
256 let started = std::time::Instant::now();
257 match self.manager.call_tool(server, tool, arguments).await {
258 Ok(text) => {
259 if let Some(record) = &self.invocation_recorder {
260 record(server, tool, true, started.elapsed().as_millis() as u64);
261 }
262 if text.is_empty() {
263 ToolOutput::ok("(empty result)")
264 } else {
265 ToolOutput::ok(self.manager.redact(text))
266 }
267 }
268 Err(e) => {
269 if let Some(record) = &self.invocation_recorder {
270 record(server, tool, false, started.elapsed().as_millis() as u64);
271 }
272 ToolOutput::error(format!(
273 "mcp call failed: {}",
274 self.manager.redact(e.to_string())
275 ))
276 }
277 }
278 }
279}
280

Indexing the workspace…