1use rmcp::ErrorData as McpError;
4use rmcp::handler::server::wrapper::Parameters;
5use rmcp::model::*;
6use rmcp::schemars;
7use rmcp::{ServerHandler, tool, tool_handler, tool_router};
8
9use serde::Deserialize;
10use std::collections::HashMap;
11use std::sync::Arc;
12
13use crate::Database;
14use crate::types::AttributeValue;
15
16use super::errors::to_tool_error;
17use crate::snapshots;
18
19#[derive(Clone, Debug)]
25pub struct McpConfig {
26 pub read_only: bool,
28 pub max_items: Option<usize>,
30 pub max_size_bytes: Option<usize>,
32 pub data_model: Option<crate::schema::DataModel>,
34 pub data_model_summary_limit: usize,
36 pub import: Option<ImportReport>,
39}
40
41impl Default for McpConfig {
42 fn default() -> Self {
43 Self {
44 read_only: false,
45 max_items: None,
46 max_size_bytes: None,
47 data_model: None,
48 data_model_summary_limit: 20,
49 import: None,
50 }
51 }
52}
53
54#[derive(Clone, Copy, Debug, PartialEq, Eq)]
57pub enum ImportConcern {
58 Exposure,
60 Caution,
62}
63
64impl ImportConcern {
65 fn as_str(self) -> &'static str {
66 match self {
67 Self::Exposure => "exposure",
68 Self::Caution => "caution",
69 }
70 }
71}
72
73#[derive(Clone, Debug, PartialEq, Eq)]
75pub struct ImportNotice {
76 pub concern: ImportConcern,
77 pub message: String,
78}
79
80#[derive(Clone, Debug, Default, PartialEq, Eq)]
85pub struct ImportReport {
86 pub tables: usize,
88 pub items: usize,
90 pub skipped: usize,
92 pub notices: Vec<ImportNotice>,
94 pub exposures_accepted: bool,
96}
97
98impl ImportReport {
99 pub fn exposure_count(&self) -> usize {
101 self.notices
102 .iter()
103 .filter(|n| n.concern == ImportConcern::Exposure)
104 .count()
105 }
106
107 pub fn instructions_sentence(&self) -> String {
109 let notices = self.notices.len();
110 let exposures = self.exposure_count();
111 if notices == 0 {
112 return "This database was populated by dynoxide import, which raised no notices."
113 .to_string();
114 }
115 let counted = format!(
116 "This database was populated by dynoxide import; get_database_info reports {} {}",
117 notices,
118 if notices == 1 { "notice" } else { "notices" }
119 );
120 if exposures == 0 {
121 return format!("{counted}, none of them exposures.");
122 }
123 let which = if exposures == 1 {
124 format!("{exposures} is an exposure")
125 } else {
126 format!("{exposures} are exposures")
127 };
128 if self.exposures_accepted {
129 let was = if exposures == 1 { "was" } else { "were" };
130 format!("{counted}, of which {which} that {was} accepted with --accept-exposure.")
131 } else {
132 format!("{counted}, of which {which}.")
133 }
134 }
135
136 fn to_json(&self) -> serde_json::Value {
138 let notices: Vec<serde_json::Value> = self
139 .notices
140 .iter()
141 .map(|n| {
142 serde_json::json!({
143 "concern": n.concern.as_str(),
144 "message": n.message,
145 })
146 })
147 .collect();
148 serde_json::json!({
149 "tables": self.tables,
150 "items": self.items,
151 "skipped": self.skipped,
152 "notice_count": self.notices.len(),
153 "exposure_count": self.exposure_count(),
154 "exposures_accepted": self.exposures_accepted,
155 "notices": notices,
156 })
157 }
158}
159
160#[derive(Clone)]
165pub struct McpServer {
166 db: Arc<Database>,
167 config: McpConfig,
168}
169
170const RETURN_CONSUMED_CAPACITY: &str = "What consumed capacity to return: TOTAL for the \
176 operation's total, INDEXES for the total broken down by table and index, or NONE (the \
177 default) for none. Figures follow DynamoDB's charging rules and match what the DynamoDB \
178 API returns for the same request.";
179
180const RETURN_CONSUMED_CAPACITY_VALUES: [Option<&str>; 4] =
186 [Some("INDEXES"), Some("TOTAL"), Some("NONE"), None];
187
188#[derive(Debug, Deserialize, schemars::JsonSchema)]
189pub struct TableNameParam {
190 #[schemars(description = "Name of the table")]
191 pub table_name: String,
192}
193
194#[derive(Debug, Deserialize, schemars::JsonSchema)]
195pub struct DescribeTableParams {
196 #[schemars(description = "Name of the table to describe")]
197 pub table_name: String,
198
199 #[schemars(
200 description = "If true, return the full DynamoDB-format response instead of the agent-friendly format"
201 )]
202 #[serde(default)]
203 pub raw: bool,
204}
205
206#[derive(Debug, Deserialize, schemars::JsonSchema)]
207pub struct CreateTableParams {
208 #[schemars(description = "Name of the table to create (3-255 characters, [a-zA-Z0-9._-])")]
209 pub table_name: String,
210
211 #[schemars(
212 description = "Key schema as array of {attribute_name, key_type} objects. key_type is HASH or RANGE"
213 )]
214 pub key_schema: serde_json::Value,
215
216 #[schemars(
217 description = "Attribute definitions as array of {attribute_name, attribute_type} objects. attribute_type is S, N, or B"
218 )]
219 pub attribute_definitions: serde_json::Value,
220
221 #[schemars(
222 description = "Optional GSI definitions as array of {index_name, key_schema, projection} objects"
223 )]
224 pub global_secondary_indexes: Option<serde_json::Value>,
225
226 #[schemars(
227 description = "Optional LSI definitions as array of {index_name, key_schema, projection} objects. LSIs share the table's partition key."
228 )]
229 pub local_secondary_indexes: Option<serde_json::Value>,
230
231 #[schemars(
232 description = "Optional vector index definitions as array of {index_name, vector_attribute: {attribute_name}, dimensions, distance_function, projection} objects. distance_function is COSINE, EUCLIDEAN or DOT_PRODUCT. Optional search_schema is an array of {attribute_name, search_schema_element_type} with element type HASH or INLINE_FILTER. A table carrying one must use PAY_PER_REQUEST. Search it with search_vectors."
233 )]
234 pub vector_indexes: Option<serde_json::Value>,
235
236 #[schemars(description = "Optional stream specification {stream_enabled, stream_view_type}")]
237 pub stream_specification: Option<serde_json::Value>,
238
239 #[schemars(
240 description = "Optional SSE specification {Enabled: bool}. Accepted for compatibility; has no effect in dynoxide."
241 )]
242 pub sse_specification: Option<serde_json::Value>,
243
244 #[schemars(
245 description = "Optional billing mode: PROVISIONED or PAY_PER_REQUEST. Defaults to PROVISIONED."
246 )]
247 pub billing_mode: Option<String>,
248
249 #[schemars(
250 description = "Optional provisioned throughput {read_capacity_units, write_capacity_units}. Required when billing_mode is explicitly PROVISIONED; must be omitted for PAY_PER_REQUEST."
251 )]
252 pub provisioned_throughput: Option<serde_json::Value>,
253
254 #[schemars(
255 description = "Optional on-demand throughput ceilings {max_read_request_units, max_write_request_units}. Stored and reported back by describe_table; not enforced."
256 )]
257 pub on_demand_throughput: Option<serde_json::Value>,
258
259 #[schemars(
260 description = "Optional table class: STANDARD or STANDARD_INFREQUENT_ACCESS. Accepted for compatibility."
261 )]
262 pub table_class: Option<String>,
263
264 #[schemars(description = "Optional tags as array of {Key, Value} objects")]
265 pub tags: Option<serde_json::Value>,
266
267 #[schemars(description = "If true, prevents the table from being deleted. Default: false.")]
268 pub deletion_protection_enabled: Option<bool>,
269}
270
271#[derive(Debug, Deserialize, schemars::JsonSchema)]
272pub struct PutItemParams {
273 #[schemars(description = "Name of the table")]
274 pub table_name: String,
275
276 #[schemars(
277 description = "Item to put, in DynamoDB JSON format. Example: {\"pk\": {\"S\": \"user#1\"}, \"name\": {\"S\": \"Alice\"}}"
278 )]
279 pub item: serde_json::Value,
280
281 #[schemars(description = "Condition expression that must be satisfied for the put to succeed")]
282 pub condition_expression: Option<String>,
283
284 #[schemars(
285 description = "Substitution tokens for expression attribute names, e.g. {\"#n\": \"name\"}"
286 )]
287 pub expression_attribute_names: Option<HashMap<String, String>>,
288
289 #[schemars(
290 description = "Substitution tokens for expression attribute values, in DynamoDB JSON format"
291 )]
292 pub expression_attribute_values: Option<serde_json::Value>,
293
294 #[schemars(description = "What to return: NONE or ALL_OLD")]
295 pub return_values: Option<String>,
296
297 #[schemars(description = "What to return on ConditionalCheckFailedException: ALL_OLD or NONE")]
298 pub return_values_on_condition_check_failure: Option<String>,
299
300 #[schemars(
301 description = RETURN_CONSUMED_CAPACITY,
302 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
303 )]
304 #[serde(default)]
305 pub return_consumed_capacity: Option<String>,
306}
307
308#[derive(Debug, Deserialize, schemars::JsonSchema)]
309pub struct GetItemParams {
310 #[schemars(description = "Name of the table")]
311 pub table_name: String,
312
313 #[schemars(
314 description = "Primary key in DynamoDB JSON format. Example: {\"pk\": {\"S\": \"user#1\"}}"
315 )]
316 pub key: serde_json::Value,
317
318 #[schemars(description = "Projection expression to limit returned attributes")]
319 pub projection_expression: Option<String>,
320
321 #[schemars(description = "Substitution tokens for expression attribute names")]
322 pub expression_attribute_names: Option<HashMap<String, String>>,
323
324 #[schemars(description = "Whether to use strongly consistent reads")]
325 #[serde(default)]
326 pub consistent_read: Option<bool>,
327
328 #[schemars(
329 description = RETURN_CONSUMED_CAPACITY,
330 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
331 )]
332 #[serde(default)]
333 pub return_consumed_capacity: Option<String>,
334}
335
336#[derive(Debug, Deserialize, schemars::JsonSchema)]
337pub struct UpdateItemParams {
338 #[schemars(description = "Name of the table")]
339 pub table_name: String,
340
341 #[schemars(
342 description = "Primary key of the item to update, in DynamoDB JSON format. Example: {\"pk\": {\"S\": \"user#1\"}, \"sk\": {\"S\": \"profile\"}}"
343 )]
344 pub key: serde_json::Value,
345
346 #[schemars(
347 description = "Update expression, e.g. \"SET #n = :val, age = age + :inc REMOVE obsolete\""
348 )]
349 pub update_expression: Option<String>,
350
351 #[schemars(
352 description = "Condition expression that must be satisfied for the update to succeed"
353 )]
354 pub condition_expression: Option<String>,
355
356 #[schemars(description = "Substitution tokens for expression attribute names")]
357 pub expression_attribute_names: Option<HashMap<String, String>>,
358
359 #[schemars(
360 description = "Substitution tokens for expression attribute values, in DynamoDB JSON format"
361 )]
362 pub expression_attribute_values: Option<serde_json::Value>,
363
364 #[schemars(
365 description = "What to return: NONE, ALL_OLD, UPDATED_OLD, ALL_NEW, or UPDATED_NEW"
366 )]
367 pub return_values: Option<String>,
368
369 #[schemars(description = "What to return on ConditionalCheckFailedException: ALL_OLD or NONE")]
370 pub return_values_on_condition_check_failure: Option<String>,
371
372 #[schemars(
373 description = RETURN_CONSUMED_CAPACITY,
374 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
375 )]
376 #[serde(default)]
377 pub return_consumed_capacity: Option<String>,
378}
379
380#[derive(Debug, Deserialize, schemars::JsonSchema)]
381pub struct SearchVectorsParams {
382 #[schemars(description = "Name of the table holding the vector index")]
383 pub table_name: String,
384
385 #[schemars(description = "Name of the vector index to search")]
386 pub index_name: String,
387
388 #[schemars(
389 description = "Query vector as an array of DynamoDB numbers, e.g. [{\"N\": \"0.1\"}, {\"N\": \"0.2\"}]. Its length must equal the index's dimensions."
390 )]
391 pub search_vector: serde_json::Value,
392
393 #[schemars(description = "How many nearest entries to return.")]
394 pub top_k: i64,
395
396 #[schemars(
397 description = "Optional equality conditions over the index's SearchSchema attributes, e.g. \"tenant = :t\". Equality only, one condition per attribute, and mandatory when the schema declares a HASH element."
398 )]
399 pub search_condition_expression: Option<String>,
400
401 #[schemars(
402 description = "Optional projection expression. The vector attribute is excluded unless named here and projected by the index."
403 )]
404 pub projection_expression: Option<String>,
405
406 #[schemars(description = "Optional expression attribute name substitutions")]
407 pub expression_attribute_names: Option<std::collections::HashMap<String, String>>,
408
409 #[schemars(description = "Optional expression attribute values as DynamoDB-typed JSON")]
410 pub expression_attribute_values: Option<serde_json::Value>,
411
412 #[schemars(
413 description = RETURN_CONSUMED_CAPACITY,
414 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
415 )]
416 #[serde(default)]
417 pub return_consumed_capacity: Option<String>,
418}
419
420#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
421pub struct QueryParams {
422 #[schemars(description = "Name of the table to query")]
423 pub table_name: String,
424
425 #[schemars(
426 description = "Key condition expression, e.g. \"pk = :pk AND begins_with(sk, :prefix)\""
427 )]
428 pub key_condition_expression: String,
429
430 #[schemars(description = "Filter expression applied after the query")]
431 pub filter_expression: Option<String>,
432
433 #[schemars(description = "Projection expression to limit returned attributes")]
434 pub projection_expression: Option<String>,
435
436 #[schemars(description = "Substitution tokens for expression attribute names")]
437 pub expression_attribute_names: Option<HashMap<String, String>>,
438
439 #[schemars(
440 description = "Substitution tokens for expression attribute values, in DynamoDB JSON format"
441 )]
442 pub expression_attribute_values: Option<serde_json::Value>,
443
444 #[schemars(
445 description = "Whether to scan forward (true, default) or backward (false) by sort key"
446 )]
447 pub scan_index_forward: Option<bool>,
448
449 #[schemars(description = "Maximum number of items to evaluate")]
450 pub limit: Option<usize>,
451
452 #[schemars(description = "Exclusive start key for pagination, in DynamoDB JSON format")]
453 pub exclusive_start_key: Option<serde_json::Value>,
454
455 #[schemars(description = "Name of a GSI or LSI to query")]
456 pub index_name: Option<String>,
457
458 #[schemars(
459 description = "What to return: ALL_ATTRIBUTES (default), ALL_PROJECTED_ATTRIBUTES, COUNT, or SPECIFIC_ATTRIBUTES. When COUNT, the response contains only Count and ScannedCount (no items)."
460 )]
461 pub select: Option<String>,
462
463 #[schemars(
464 description = RETURN_CONSUMED_CAPACITY,
465 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
466 )]
467 #[serde(default)]
468 pub return_consumed_capacity: Option<String>,
469}
470
471#[derive(Debug, Deserialize, schemars::JsonSchema)]
472pub struct ScanParams {
473 #[schemars(description = "Name of the table to scan")]
474 pub table_name: String,
475
476 #[schemars(description = "Filter expression to apply")]
477 pub filter_expression: Option<String>,
478
479 #[schemars(description = "Projection expression to limit returned attributes")]
480 pub projection_expression: Option<String>,
481
482 #[schemars(description = "Substitution tokens for expression attribute names")]
483 pub expression_attribute_names: Option<HashMap<String, String>>,
484
485 #[schemars(
486 description = "Substitution tokens for expression attribute values, in DynamoDB JSON format"
487 )]
488 pub expression_attribute_values: Option<serde_json::Value>,
489
490 #[schemars(description = "Maximum number of items to evaluate")]
491 pub limit: Option<usize>,
492
493 #[schemars(description = "Exclusive start key for pagination, in DynamoDB JSON format")]
494 pub exclusive_start_key: Option<serde_json::Value>,
495
496 #[schemars(description = "Name of a GSI or LSI to scan")]
497 pub index_name: Option<String>,
498
499 #[schemars(
500 description = "What to return: ALL_ATTRIBUTES (default), ALL_PROJECTED_ATTRIBUTES, COUNT, or SPECIFIC_ATTRIBUTES. When COUNT, the response contains only Count and ScannedCount (no items)."
501 )]
502 pub select: Option<String>,
503
504 #[schemars(
505 description = "Segment number for parallel scan (0-based). Must be used with total_segments."
506 )]
507 pub segment: Option<u32>,
508
509 #[schemars(
510 description = "Total number of parallel scan segments (1 to 1000000). Must be used with segment."
511 )]
512 pub total_segments: Option<u32>,
513
514 #[schemars(
515 description = RETURN_CONSUMED_CAPACITY,
516 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
517 )]
518 #[serde(default)]
519 pub return_consumed_capacity: Option<String>,
520}
521
522#[derive(Debug, Deserialize, schemars::JsonSchema)]
523pub struct DeleteItemParams {
524 #[schemars(description = "Name of the table")]
525 pub table_name: String,
526
527 #[schemars(
528 description = "Primary key of the item to delete, in DynamoDB JSON format. Example: {\"pk\": {\"S\": \"user#1\"}, \"sk\": {\"S\": \"profile\"}}"
529 )]
530 pub key: serde_json::Value,
531
532 #[schemars(
533 description = "Condition expression that must be satisfied for the delete to succeed"
534 )]
535 pub condition_expression: Option<String>,
536
537 #[schemars(description = "Substitution tokens for expression attribute names")]
538 pub expression_attribute_names: Option<HashMap<String, String>>,
539
540 #[schemars(
541 description = "Substitution tokens for expression attribute values, in DynamoDB JSON format"
542 )]
543 pub expression_attribute_values: Option<serde_json::Value>,
544
545 #[schemars(description = "What to return: NONE or ALL_OLD")]
546 pub return_values: Option<String>,
547
548 #[schemars(description = "What to return on ConditionalCheckFailedException: ALL_OLD or NONE")]
549 pub return_values_on_condition_check_failure: Option<String>,
550
551 #[schemars(
552 description = RETURN_CONSUMED_CAPACITY,
553 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
554 )]
555 #[serde(default)]
556 pub return_consumed_capacity: Option<String>,
557}
558
559#[derive(Debug, Deserialize, schemars::JsonSchema)]
560pub struct BatchWriteItemParams {
561 #[schemars(
562 description = "Map of table name to array of write requests. Each request is either {put_request: {item: ...}} or {delete_request: {key: ...}}. Max 25 operations total."
563 )]
564 pub request_items: serde_json::Value,
565
566 #[schemars(
567 description = RETURN_CONSUMED_CAPACITY,
568 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
569 )]
570 #[serde(default)]
571 pub return_consumed_capacity: Option<String>,
572}
573
574#[derive(Debug, Deserialize, schemars::JsonSchema)]
575pub struct BatchGetItemParams {
576 #[schemars(
577 description = "Map of table name to {keys: [...], projection_expression?, expression_attribute_names?}. Max 100 keys total."
578 )]
579 pub request_items: serde_json::Value,
580
581 #[schemars(
582 description = RETURN_CONSUMED_CAPACITY,
583 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
584 )]
585 #[serde(default)]
586 pub return_consumed_capacity: Option<String>,
587}
588
589#[derive(Debug, Deserialize, schemars::JsonSchema)]
590pub struct ExecutePartiqlParams {
591 #[schemars(
592 description = "PartiQL statement to execute. Supports SELECT (with LIMIT, nested path projections, and NextToken paging), INSERT (with IF NOT EXISTS, parameter placeholders), UPDATE (with SET arithmetic, REMOVE, nested paths), DELETE. WHERE supports BETWEEN, IN, CONTAINS, IS MISSING, IS NOT MISSING, AND, OR, NOT, parenthesised grouping up to 64 levels deep, and nested paths; AND binds tighter than OR, and a NOT over a group is applied by De Morgan. An index is named in the FROM clause as \"table\".\"index\", which serves the read from that GSI or LSI on SELECT only; a write statement rejects the qualifier."
593 )]
594 pub statement: String,
595
596 #[schemars(description = "Positional parameters for the statement, in DynamoDB JSON format")]
597 pub parameters: Option<serde_json::Value>,
598
599 #[schemars(
600 description = "Whether to use strongly consistent reads. Does not change which rows come back, but it doubles the rate a SELECT is charged at and it makes a SELECT qualified by a GSI a rejection, because a GSI cannot serve one."
601 )]
602 #[serde(default)]
603 pub consistent_read: Option<bool>,
604
605 #[schemars(
606 description = "Maximum number of items to evaluate for SELECT statements (not necessarily the number returned); a filtered page can come back short or empty and still carry a NextToken"
607 )]
608 pub limit: Option<usize>,
609
610 #[schemars(
611 description = "Continuation token from a previous SELECT response, to fetch the next page. Pass back the NextToken the last call returned; its absence means there are no more pages. The statement must stay identical character for character across pages, and so must the parameters; a token replayed with any difference is rejected with a ValidationException. A statement with a list or map parameter cannot be paged past its first page, as on DynamoDB."
612 )]
613 pub next_token: Option<String>,
614
615 #[schemars(
616 description = RETURN_CONSUMED_CAPACITY,
617 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
618 )]
619 #[serde(default)]
620 pub return_consumed_capacity: Option<String>,
621
622 #[schemars(
623 description = "What to return when an UPDATE or DELETE fails its WHERE condition: ALL_OLD hands back the item the condition was evaluated against as the error's item; NONE, the default, does not. Any other value is rejected with a ValidationException.",
624 extend("enum" = ["ALL_OLD", "NONE"])
625 )]
626 #[serde(default)]
627 pub return_values_on_condition_check_failure: Option<String>,
628}
629
630#[derive(Debug, Deserialize, schemars::JsonSchema)]
631pub struct CreateSnapshotParams {
632 #[schemars(
633 description = "Optional name for the snapshot. If omitted, a timestamped name is generated."
634 )]
635 pub name: Option<String>,
636}
637
638#[derive(Debug, Deserialize, schemars::JsonSchema)]
639pub struct RestoreSnapshotParams {
640 #[schemars(
641 description = "Snapshot name to restore (e.g., 'my-snapshot' or 'pre-delete-Users-20260308T143022Z'). Use list_snapshots to see available names."
642 )]
643 pub name: String,
644}
645
646#[derive(Debug, Deserialize, schemars::JsonSchema)]
647pub struct DeleteSnapshotParams {
648 #[schemars(
649 description = "Snapshot name to delete (e.g., 'my-snapshot'). Use list_snapshots to see available names."
650 )]
651 pub name: String,
652}
653
654#[derive(Debug, Deserialize, schemars::JsonSchema)]
655pub struct UpdateTableParams {
656 #[schemars(description = "Name of the table to update")]
657 pub table_name: String,
658
659 #[schemars(
660 description = "Attribute definitions for new GSI key attributes, as array of {attribute_name, attribute_type} objects. Only attributes used by a key schema are kept; any others are dropped, and a redeclared attribute keeps its existing type, matching DynamoDB"
661 )]
662 pub attribute_definitions: Option<serde_json::Value>,
663
664 #[schemars(
665 description = "GSI updates as array of {Create?: {index_name, key_schema, projection}, Delete?: {index_name}} objects"
666 )]
667 pub global_secondary_index_updates: Option<serde_json::Value>,
668
669 #[schemars(
670 description = "Vector index updates as array of {create} or {delete} objects. A create carries the same shape as a create_table vector index; a delete carries {index_name}. One index per call. A created index is not searchable straight away: it reports index_status CREATING, then ACTIVE while still refusing searches. The table cannot be deleted while the index reports CREATING, which is the earlier of those two and lifts before searches start working. A delete of the index is refused for a short allocation window after the create, with an answer naming the phase to retry in, and accepted from then on."
671 )]
672 pub vector_index_updates: Option<serde_json::Value>,
673
674 #[schemars(description = "Stream specification {stream_enabled, stream_view_type}")]
675 pub stream_specification: Option<serde_json::Value>,
676
677 #[schemars(description = "Enable or disable deletion protection on the table")]
678 pub deletion_protection_enabled: Option<bool>,
679
680 #[schemars(
681 description = "Switch billing mode: PROVISIONED or PAY_PER_REQUEST. Switching to PROVISIONED requires provisioned_throughput."
682 )]
683 pub billing_mode: Option<String>,
684
685 #[schemars(
686 description = "New provisioned throughput {read_capacity_units, write_capacity_units}. Not allowed while the table is PAY_PER_REQUEST unless billing_mode switches to PROVISIONED."
687 )]
688 pub provisioned_throughput: Option<serde_json::Value>,
689
690 #[schemars(description = "Change table class: STANDARD or STANDARD_INFREQUENT_ACCESS")]
691 pub table_class: Option<String>,
692
693 #[schemars(
694 description = "New on-demand throughput ceilings {max_read_request_units, max_write_request_units}. Stored and reported back by describe_table; not enforced."
695 )]
696 pub on_demand_throughput: Option<serde_json::Value>,
697}
698
699#[derive(Debug, Deserialize, schemars::JsonSchema)]
700pub struct TransactWriteItemsParams {
701 #[schemars(
702 description = "Array of transact items (max 100). Each item is one of: {Put: {TableName, Item, ...}}, {Update: {TableName, Key, UpdateExpression, ...}}, {Delete: {TableName, Key, ...}}, {ConditionCheck: {TableName, Key, ConditionExpression, ...}}"
703 )]
704 pub transact_items: serde_json::Value,
705
706 #[schemars(
707 description = "Idempotency token (max 36 chars). Same token returns cached result."
708 )]
709 pub client_request_token: Option<String>,
710
711 #[schemars(
712 description = RETURN_CONSUMED_CAPACITY,
713 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
714 )]
715 #[serde(default)]
716 pub return_consumed_capacity: Option<String>,
717}
718
719#[derive(Debug, Deserialize, schemars::JsonSchema)]
720pub struct TransactGetItemsParams {
721 #[schemars(
722 description = "Array of get items (max 100). Each item is: {Get: {TableName, Key, ProjectionExpression?, ExpressionAttributeNames?}}"
723 )]
724 pub transact_items: serde_json::Value,
725
726 #[schemars(
727 description = RETURN_CONSUMED_CAPACITY,
728 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
729 )]
730 #[serde(default)]
731 pub return_consumed_capacity: Option<String>,
732}
733
734#[derive(Debug, Deserialize, schemars::JsonSchema)]
735pub struct BatchExecutePartiqlParams {
736 #[schemars(
737 description = "Array of PartiQL statements. Each is {Statement: \"...\", Parameters?: [...], ConsistentRead?: true|false, ReturnValuesOnConditionCheckFailure?: \"ALL_OLD\"|\"NONE\"}. ConsistentRead is per member and sets the rate that member's read is charged at rather than which rows come back. ReturnValuesOnConditionCheckFailure ALL_OLD returns the item a failed UPDATE or DELETE condition was evaluated against on that member's Error. A batch is all-read or all-write, and a SELECT member must name the table's primary key and may not name an index."
738 )]
739 pub statements: serde_json::Value,
740
741 #[schemars(
742 description = RETURN_CONSUMED_CAPACITY,
743 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
744 )]
745 #[serde(default)]
746 pub return_consumed_capacity: Option<String>,
747}
748
749#[derive(Debug, Deserialize, schemars::JsonSchema)]
750pub struct ExecuteTransactionPartiqlParams {
751 #[schemars(
752 description = "Array of PartiQL statements (max 100). Each is {Statement: \"...\", Parameters?: [...], ReturnValuesOnConditionCheckFailure?: \"ALL_OLD\"|\"NONE\"}. All succeed or all fail atomically. ALL_OLD returns the item a failed UPDATE or DELETE condition was evaluated against on that statement's cancellation reason."
753 )]
754 pub transact_statements: serde_json::Value,
755 #[schemars(description = "Optional idempotency token for the transaction")]
756 pub client_request_token: Option<String>,
757 #[schemars(
758 description = RETURN_CONSUMED_CAPACITY,
759 extend("enum" = RETURN_CONSUMED_CAPACITY_VALUES)
760 )]
761 #[serde(default)]
762 pub return_consumed_capacity: Option<String>,
763}
764
765#[derive(Debug, Deserialize, schemars::JsonSchema)]
766pub struct UpdateTimeToLiveParams {
767 #[schemars(description = "Name of the table")]
768 pub table_name: String,
769
770 #[schemars(
771 description = "TTL specification: {Enabled: true/false, AttributeName: \"ttl_attr\"}"
772 )]
773 pub time_to_live_specification: serde_json::Value,
774}
775
776#[derive(Debug, Deserialize, schemars::JsonSchema)]
777pub struct DescribeTimeToLiveParams {
778 #[schemars(description = "Name of the table")]
779 pub table_name: String,
780}
781
782#[derive(Debug, Deserialize, schemars::JsonSchema)]
783pub struct ListStreamsParams {
784 #[schemars(description = "Filter streams by table name")]
785 pub table_name: Option<String>,
786
787 #[schemars(description = "Maximum number of streams to return")]
788 pub limit: Option<usize>,
789}
790
791#[derive(Debug, Deserialize, schemars::JsonSchema)]
792pub struct DescribeStreamParams {
793 #[schemars(description = "ARN of the stream to describe")]
794 pub stream_arn: String,
795
796 #[schemars(description = "Maximum number of shards to return")]
797 pub limit: Option<usize>,
798}
799
800#[derive(Debug, Deserialize, schemars::JsonSchema)]
801pub struct GetShardIteratorParams {
802 #[schemars(description = "ARN of the stream")]
803 pub stream_arn: String,
804
805 #[schemars(description = "ID of the shard")]
806 pub shard_id: String,
807
808 #[schemars(
809 description = "Iterator type: TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, or AFTER_SEQUENCE_NUMBER"
810 )]
811 pub shard_iterator_type: String,
812
813 #[schemars(
814 description = "Sequence number, 21 to 40 digits (required for AT/AFTER_SEQUENCE_NUMBER, refused for TRIM_HORIZON and LATEST)"
815 )]
816 pub sequence_number: Option<String>,
817}
818
819#[derive(Debug, Deserialize, schemars::JsonSchema)]
820pub struct GetRecordsParams {
821 #[schemars(description = "Shard iterator from get_shard_iterator")]
822 pub shard_iterator: String,
823
824 #[schemars(description = "Maximum number of records to return")]
825 pub limit: Option<usize>,
826}
827
828#[derive(Debug, Deserialize, schemars::JsonSchema)]
829pub struct TagResourceParams {
830 #[schemars(description = "ARN of the resource (table ARN)")]
831 pub resource_arn: String,
832
833 #[schemars(description = "Array of tags: [{Key: \"env\", Value: \"prod\"}, ...]")]
834 pub tags: serde_json::Value,
835}
836
837#[derive(Debug, Deserialize, schemars::JsonSchema)]
838pub struct UntagResourceParams {
839 #[schemars(description = "ARN of the resource (table ARN)")]
840 pub resource_arn: String,
841
842 #[schemars(description = "Array of tag keys to remove: [\"env\", \"team\"]")]
843 pub tag_keys: Vec<String>,
844}
845
846#[derive(Debug, Deserialize, schemars::JsonSchema)]
847pub struct ListTagsOfResourceParams {
848 #[schemars(description = "ARN of the resource (table ARN)")]
849 pub resource_arn: String,
850}
851
852#[derive(Debug, Deserialize, schemars::JsonSchema)]
853pub struct BulkPutItemsParams {
854 #[schemars(description = "Name of the table")]
855 pub table_name: String,
856
857 #[schemars(
858 description = "Array of items in DynamoDB JSON format. Example: [{\"pk\": {\"S\": \"user#1\"}, \"name\": {\"S\": \"Alice\"}}, ...]. Maximum 10,000 items per call."
859 )]
860 pub items: Vec<serde_json::Value>,
861
862 #[schemars(
863 description = "Whether to record DynamoDB Stream events for the imported items. Default: false."
864 )]
865 #[serde(default)]
866 pub record_streams: bool,
867}
868
869fn decode_dynamo_map(
879 val: serde_json::Value,
880 field_name: &str,
881) -> Result<HashMap<String, AttributeValue>, crate::errors::DynoxideError> {
882 serde_json::from_value(val).map_err(|e| {
883 let msg = e.to_string();
884 crate::serde_errors::classify_marked_serde_error(&msg).unwrap_or_else(|| {
885 crate::errors::DynoxideError::ValidationException(format!(
886 "Invalid DynamoDB JSON in '{field_name}': {msg}"
887 ))
888 })
889 })
890}
891
892fn parse_dynamo_map(
898 val: serde_json::Value,
899 field_name: &str,
900 resolve: fn(crate::DynoxideError) -> crate::DynoxideError,
901) -> Result<HashMap<String, AttributeValue>, CallToolResult> {
902 decode_dynamo_map(val, field_name).map_err(|e| to_tool_error(resolve(e)))
903}
904
905fn parse_optional_dynamo_map(
907 val: Option<serde_json::Value>,
908 field_name: &str,
909 resolve: fn(crate::DynoxideError) -> crate::DynoxideError,
910) -> Result<Option<HashMap<String, AttributeValue>>, CallToolResult> {
911 match val {
912 Some(v) => parse_dynamo_map(v, field_name, resolve).map(Some),
913 None => Ok(None),
914 }
915}
916
917fn canonicalise_vector_index_updates(val: &serde_json::Value) -> serde_json::Value {
930 let Some(arr) = val.as_array() else {
931 return val.clone();
932 };
933 serde_json::Value::Array(
934 arr.iter()
935 .map(|entry| {
936 let Some(obj) = entry.as_object() else {
937 return entry.clone();
938 };
939 let mut out = serde_json::Map::new();
940 if let Some(c) = obj.get("Create").or_else(|| obj.get("create")) {
946 out.insert("Create".to_string(), canonicalise_member_names(c));
947 }
948 if let Some(d) = obj.get("Delete").or_else(|| obj.get("delete")) {
949 out.insert("Delete".to_string(), canonicalise_member_names(d));
950 }
951 if out.is_empty() {
952 return entry.clone();
953 }
954 serde_json::Value::Object(out)
955 })
956 .collect(),
957 )
958}
959
960fn wire_member_name(key: &str) -> &str {
967 match key {
968 "index_name" => "IndexName",
969 "vector_attribute" => "VectorAttribute",
970 "search_schema" => "SearchSchema",
971 "projection" => "Projection",
972 "dimensions" => "Dimensions",
973 "distance_function" => "DistanceFunction",
974 "attribute_name" => "AttributeName",
975 "search_schema_element_type" => "SearchSchemaElementType",
976 "projection_type" => "ProjectionType",
977 "non_key_attributes" => "NonKeyAttributes",
978 other => other,
979 }
980}
981
982fn canonicalise_member_names(val: &serde_json::Value) -> serde_json::Value {
991 match val {
992 serde_json::Value::Object(obj) => serde_json::Value::Object(
993 obj.iter()
994 .map(|(k, v)| {
995 (
996 wire_member_name(k).to_string(),
997 canonicalise_member_names(v),
998 )
999 })
1000 .collect(),
1001 ),
1002 serde_json::Value::Array(arr) => {
1003 serde_json::Value::Array(arr.iter().map(canonicalise_member_names).collect())
1004 }
1005 other => other.clone(),
1006 }
1007}
1008
1009fn tool_validation_error(error_type: &str, message: &str) -> CallToolResult {
1012 CallToolResult::error(vec![ContentBlock::text(
1013 serde_json::json!({
1014 "error_type": error_type,
1015 "message": message,
1016 "retryable": false,
1017 })
1018 .to_string(),
1019 )])
1020}
1021
1022fn refuse_bad_capacity_mode(mode: Option<&str>) -> Option<CallToolResult> {
1027 crate::validation::reject_bad_capacity_mode(mode)
1028 .err()
1029 .map(to_tool_error)
1030}
1031
1032fn parse_json_param<T: serde::de::DeserializeOwned>(
1041 val: serde_json::Value,
1042 param_name: &str,
1043) -> Result<T, CallToolResult> {
1044 serde_json::from_value(val).map_err(|e| {
1045 let msg = e.to_string();
1046 match crate::serde_errors::classify_marked_serde_error(&msg) {
1047 Some(err) => to_tool_error(crate::validation::strip_request_validation_tag(err)),
1048 None => tool_validation_error(
1049 &format!("Invalid{}", capitalize_first(param_name)),
1050 &format!(
1051 "Invalid {param_name}: {}",
1052 crate::serde_errors::clean_serde_message(&msg)
1053 ),
1054 ),
1055 }
1056 })
1057}
1058
1059fn parse_optional_json_param<T: serde::de::DeserializeOwned>(
1064 value: Option<serde_json::Value>,
1065 error_type: &str,
1066 param_name: &str,
1067) -> Result<Option<T>, CallToolResult> {
1068 value
1069 .map(|v| {
1070 serde_json::from_value(v).map_err(|e| {
1071 tool_validation_error(error_type, &format!("Invalid {param_name}: {e}"))
1072 })
1073 })
1074 .transpose()
1075}
1076
1077fn capitalize_first(s: &str) -> String {
1078 let mut chars = s.chars();
1079 match chars.next() {
1080 None => String::new(),
1081 Some(c) => c.to_uppercase().collect::<String>() + chars.as_str(),
1082 }
1083}
1084
1085fn json_result(value: impl serde::Serialize) -> Result<CallToolResult, McpError> {
1086 let json =
1087 serde_json::to_value(value).map_err(|e| McpError::internal_error(e.to_string(), None))?;
1088 Ok(CallToolResult::success(vec![ContentBlock::text(
1089 json.to_string(),
1090 )]))
1091}
1092
1093#[tool_router]
1098impl McpServer {
1099 pub fn new(db: Arc<Database>) -> Self {
1100 Self::with_config(db, McpConfig::default())
1101 }
1102
1103 pub fn with_config(db: Arc<Database>, config: McpConfig) -> Self {
1104 Self { db, config }
1105 }
1106
1107 fn reject_if_read_only(&self, tool_name: &str) -> Option<CallToolResult> {
1109 if self.config.read_only {
1110 Some(CallToolResult::error(vec![ContentBlock::text(
1111 serde_json::json!({
1112 "error_type": "AccessDeniedException",
1113 "message": format!("Tool '{tool_name}' is disabled: server is in read-only mode (--read-only)"),
1114 "retryable": false,
1115 })
1116 .to_string(),
1117 )]))
1118 } else {
1119 None
1120 }
1121 }
1122
1123 fn apply_item_limit(&self, user_limit: Option<usize>) -> Option<usize> {
1125 match (user_limit, self.config.max_items) {
1126 (Some(u), Some(m)) => Some(u.min(m)),
1127 (None, Some(m)) => Some(m),
1128 (Some(u), None) => Some(u),
1129 (None, None) => None,
1130 }
1131 }
1132
1133 fn json_result_checked(&self, value: &serde_json::Value) -> Result<CallToolResult, McpError> {
1139 let serialized =
1140 serde_json::to_vec(value).map_err(|e| McpError::internal_error(e.to_string(), None))?;
1141 if let Some(max_bytes) = self.config.max_size_bytes
1142 && serialized.len() > max_bytes
1143 {
1144 let error_json = serde_json::json!({
1145 "error_type": "ResponseSizeLimitExceeded",
1146 "message": format!(
1147 "Response size ({} bytes) exceeds --max-size-bytes limit ({} bytes). \
1148 Use a smaller `limit` parameter or add a projection_expression to reduce response size.",
1149 serialized.len(),
1150 max_bytes
1151 ),
1152 "retryable": false,
1153 });
1154 return Ok(CallToolResult::error(vec![ContentBlock::text(
1155 error_json.to_string(),
1156 )]));
1157 }
1158 let text = unsafe { String::from_utf8_unchecked(serialized) };
1160 Ok(CallToolResult::success(vec![ContentBlock::text(text)]))
1161 }
1162
1163 #[tool(
1164 description = "List all DynamoDB tables in the database. Returns table names as an array."
1165 )]
1166 fn list_tables(&self) -> Result<CallToolResult, McpError> {
1167 let request = crate::actions::list_tables::ListTablesRequest::default();
1168 match self.db.list_tables(request) {
1169 Ok(resp) => json_result(&resp),
1170 Err(err) => Ok(to_tool_error(err)),
1171 }
1172 }
1173
1174 #[tool(
1175 description = "Get detailed schema information for a table including key schema, GSIs, LSIs, item count, size, stream config, and TTL settings. Returns an agent-friendly flattened format by default; set raw=true for the full DynamoDB-format response."
1176 )]
1177 fn describe_table(
1178 &self,
1179 Parameters(params): Parameters<DescribeTableParams>,
1180 ) -> Result<CallToolResult, McpError> {
1181 let request = crate::actions::describe_table::DescribeTableRequest {
1182 table_name: params.table_name,
1183 };
1184 match self.db.describe_table(request) {
1185 Ok(resp) => {
1186 if params.raw {
1187 json_result(&resp)
1188 } else {
1189 let mut flattened = flatten_table_description(&resp.table);
1190 enrich_with_ttl(&mut flattened, &self.db, &resp.table.table_name);
1191 json_result(flattened)
1192 }
1193 }
1194 Err(err) => Ok(to_tool_error(err)),
1195 }
1196 }
1197
1198 #[tool(
1199 description = "[WRITE] Create a new DynamoDB table with specified key schema and optional GSIs and LSIs. Supports BillingMode, ProvisionedThroughput, OnDemandThroughput, SSESpecification, TableClass, Tags, and DeletionProtectionEnabled. The table is available immediately after creation."
1200 )]
1201 fn create_table(
1202 &self,
1203 Parameters(params): Parameters<CreateTableParams>,
1204 ) -> Result<CallToolResult, McpError> {
1205 if let Some(err) = self.reject_if_read_only("create_table") {
1206 return Ok(err);
1207 }
1208 let key_schema = match parse_json_param(params.key_schema, "key_schema") {
1209 Ok(v) => v,
1210 Err(err) => return Ok(err),
1211 };
1212 let attribute_definitions =
1213 match parse_json_param(params.attribute_definitions, "attribute_definitions") {
1214 Ok(v) => v,
1215 Err(err) => return Ok(err),
1216 };
1217 let global_secondary_indexes = match params
1218 .global_secondary_indexes
1219 .map(serde_json::from_value)
1220 .transpose()
1221 {
1222 Ok(v) => v,
1223 Err(e) => {
1224 return Ok(tool_validation_error(
1225 "InvalidGSI",
1226 &format!("Invalid global_secondary_indexes: {e}"),
1227 ));
1228 }
1229 };
1230 let raw_wire_vixs = params
1237 .vector_indexes
1238 .as_ref()
1239 .map(canonicalise_member_names);
1240 if let Some(msg) =
1241 crate::actions::create_table::vector_indexes_request_model_error(&raw_wire_vixs)
1242 {
1243 return Ok(tool_validation_error("ValidationException", &msg));
1244 }
1245 let vector_indexes: Option<Vec<crate::types::VectorIndex>> = match params
1248 .vector_indexes
1249 .as_ref()
1250 .map(crate::actions::create_table::normalise_vix_dimensions)
1251 .map(serde_json::from_value)
1252 .transpose()
1253 {
1254 Ok(v) => v,
1255 Err(e) => {
1256 return Ok(tool_validation_error(
1257 "InvalidVectorIndexes",
1258 &format!("Invalid vector_indexes: {e}"),
1259 ));
1260 }
1261 };
1262 let local_secondary_indexes = match params
1263 .local_secondary_indexes
1264 .map(serde_json::from_value)
1265 .transpose()
1266 {
1267 Ok(v) => v,
1268 Err(e) => {
1269 return Ok(tool_validation_error(
1270 "InvalidLSI",
1271 &format!("Invalid local_secondary_indexes: {e}"),
1272 ));
1273 }
1274 };
1275 let stream_specification = match params
1276 .stream_specification
1277 .map(serde_json::from_value)
1278 .transpose()
1279 {
1280 Ok(v) => v,
1281 Err(e) => {
1282 return Ok(tool_validation_error(
1283 "InvalidStreamSpec",
1284 &format!("Invalid stream_specification: {e}"),
1285 ));
1286 }
1287 };
1288 let sse_specification = match params
1289 .sse_specification
1290 .map(serde_json::from_value)
1291 .transpose()
1292 {
1293 Ok(v) => v,
1294 Err(e) => {
1295 return Ok(tool_validation_error(
1296 "InvalidSSESpec",
1297 &format!("Invalid sse_specification: {e}"),
1298 ));
1299 }
1300 };
1301 let tags = match params.tags.map(serde_json::from_value).transpose() {
1302 Ok(v) => v,
1303 Err(e) => {
1304 return Ok(tool_validation_error(
1305 "InvalidTags",
1306 &format!("Invalid tags: {e}"),
1307 ));
1308 }
1309 };
1310 let provisioned_throughput = match params
1311 .provisioned_throughput
1312 .map(serde_json::from_value)
1313 .transpose()
1314 {
1315 Ok(v) => v,
1316 Err(e) => {
1317 return Ok(tool_validation_error(
1318 "InvalidProvisionedThroughput",
1319 &format!("Invalid provisioned_throughput: {e}"),
1320 ));
1321 }
1322 };
1323 let on_demand_throughput = match parse_optional_json_param(
1324 params.on_demand_throughput,
1325 "InvalidOnDemandThroughput",
1326 "on_demand_throughput",
1327 ) {
1328 Ok(v) => v,
1329 Err(e) => return Ok(e),
1330 };
1331
1332 let request = crate::actions::create_table::CreateTableRequest {
1333 table_name: params.table_name,
1334 key_schema,
1335 attribute_definitions,
1336 global_secondary_indexes,
1337 local_secondary_indexes,
1338 vector_indexes,
1339 stream_specification,
1340 sse_specification,
1341 billing_mode: params.billing_mode,
1342 provisioned_throughput,
1343 on_demand_throughput,
1344 table_class: params.table_class,
1345 tags,
1346 deletion_protection_enabled: params.deletion_protection_enabled,
1347 };
1348 match self.db.create_table(request) {
1349 Ok(resp) => json_result(&resp),
1350 Err(err) => Ok(to_tool_error(err)),
1351 }
1352 }
1353
1354 #[tool(
1355 description = "[DESTRUCTIVE] Delete a table and all its data. An auto-snapshot is created before deletion for recovery via restore_snapshot."
1356 )]
1357 fn delete_table(
1358 &self,
1359 Parameters(params): Parameters<TableNameParam>,
1360 ) -> Result<CallToolResult, McpError> {
1361 if let Some(err) = self.reject_if_read_only("delete_table") {
1362 return Ok(err);
1363 }
1364 let snap_info = match snapshots::auto_snapshot(&self.db, ¶ms.table_name) {
1366 Ok(info) => info,
1367 Err(e) => {
1368 return Ok(CallToolResult::error(vec![ContentBlock::text(
1369 serde_json::json!({
1370 "error_type": "SnapshotFailed",
1371 "message": format!(
1372 "Auto-snapshot failed, delete_table aborted to prevent data loss: {e}"
1373 ),
1374 })
1375 .to_string(),
1376 )]));
1377 }
1378 };
1379
1380 let request = crate::actions::delete_table::DeleteTableRequest {
1381 table_name: params.table_name,
1382 };
1383 match self.db.delete_table(request) {
1384 Ok(resp) => {
1385 let mut result = serde_json::to_value(&resp)
1386 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1387 result["_auto_snapshot"] = serde_json::json!({
1388 "name": snap_info.name,
1389 "message": "Auto-snapshot created before deletion. Use restore_snapshot to recover.",
1390 });
1391 json_result(&result)
1392 }
1393 Err(err) => Ok(to_tool_error(err)),
1394 }
1395 }
1396
1397 #[tool(
1398 description = "[WRITE] Insert or replace an item in a table. Supports conditional writes via condition_expression. Empty strings, empty sets, and invalid numbers are rejected. Sets are deduplicated. Unused ExpressionAttributeNames/Values are rejected."
1399 )]
1400 fn put_item(
1401 &self,
1402 Parameters(params): Parameters<PutItemParams>,
1403 ) -> Result<CallToolResult, McpError> {
1404 if let Some(err) = self.reject_if_read_only("put_item") {
1405 return Ok(err);
1406 }
1407 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1408 return Ok(err);
1409 }
1410 let item = match parse_dynamo_map(
1411 params.item,
1412 "item",
1413 crate::validation::envelope_request_validation,
1414 ) {
1415 Ok(v) => v,
1416 Err(err) => return Ok(err),
1417 };
1418 let expression_attribute_values = match parse_optional_dynamo_map(
1419 params.expression_attribute_values,
1420 "expression_attribute_values",
1421 crate::validation::envelope_request_validation,
1422 ) {
1423 Ok(v) => v,
1424 Err(err) => return Ok(err),
1425 };
1426
1427 let request = crate::actions::put_item::PutItemRequest {
1428 table_name: params.table_name,
1429 item,
1430 condition_expression: params.condition_expression,
1431 expression_attribute_names: params.expression_attribute_names,
1432 expression_attribute_values,
1433 return_values: params.return_values,
1434 return_values_on_condition_check_failure: params
1435 .return_values_on_condition_check_failure,
1436 return_consumed_capacity: params.return_consumed_capacity,
1437 ..Default::default()
1438 };
1439 match self.db.put_item(request) {
1440 Ok(resp) => json_result(&resp),
1441 Err(err) => Ok(to_tool_error(err)),
1442 }
1443 }
1444
1445 #[tool(description = "Retrieve a single item by its primary key.")]
1446 fn get_item(
1447 &self,
1448 Parameters(params): Parameters<GetItemParams>,
1449 ) -> Result<CallToolResult, McpError> {
1450 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1451 return Ok(err);
1452 }
1453 let key = match parse_dynamo_map(
1454 params.key,
1455 "key",
1456 crate::validation::strip_request_validation_tag,
1457 ) {
1458 Ok(v) => v,
1459 Err(err) => return Ok(err),
1460 };
1461
1462 let request = crate::actions::get_item::GetItemRequest {
1463 table_name: params.table_name,
1464 key,
1465 projection_expression: params.projection_expression,
1466 expression_attribute_names: params.expression_attribute_names,
1467 consistent_read: params.consistent_read,
1468 return_consumed_capacity: params.return_consumed_capacity,
1469 ..Default::default()
1470 };
1471 match self.db.get_item(request) {
1472 Ok(resp) => json_result(&resp),
1473 Err(err) => Ok(to_tool_error(err)),
1474 }
1475 }
1476
1477 #[tool(
1478 description = "[WRITE] Update an item's attributes using an update expression. Supports SET, REMOVE, ADD, DELETE actions and conditional updates. REMOVE/ADD/DELETE on key attributes are rejected. Arithmetic uses arbitrary-precision decimal."
1479 )]
1480 fn update_item(
1481 &self,
1482 Parameters(params): Parameters<UpdateItemParams>,
1483 ) -> Result<CallToolResult, McpError> {
1484 if let Some(err) = self.reject_if_read_only("update_item") {
1485 return Ok(err);
1486 }
1487 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1488 return Ok(err);
1489 }
1490 let key = match parse_dynamo_map(
1491 params.key,
1492 "key",
1493 crate::validation::envelope_request_validation,
1494 ) {
1495 Ok(v) => v,
1496 Err(err) => return Ok(err),
1497 };
1498 let expression_attribute_values = match parse_optional_dynamo_map(
1499 params.expression_attribute_values,
1500 "expression_attribute_values",
1501 crate::validation::envelope_request_validation,
1502 ) {
1503 Ok(v) => v,
1504 Err(err) => return Ok(err),
1505 };
1506
1507 let request = crate::actions::update_item::UpdateItemRequest {
1508 table_name: params.table_name,
1509 key,
1510 update_expression: params.update_expression,
1511 condition_expression: params.condition_expression,
1512 expression_attribute_names: params.expression_attribute_names,
1513 expression_attribute_values,
1514 return_values: params.return_values,
1515 return_values_on_condition_check_failure: params
1516 .return_values_on_condition_check_failure,
1517 return_consumed_capacity: params.return_consumed_capacity,
1518 ..Default::default()
1519 };
1520 match self.db.update_item(request) {
1521 Ok(resp) => json_result(&resp),
1522 Err(err) => Ok(to_tool_error(err)),
1523 }
1524 }
1525
1526 #[tool(
1527 description = "[READ] Search a vector index for the nearest entries to a query vector. Exact brute-force KNN over the whole index, so the top-k is the true top-k. COSINE and EUCLIDEAN score as distances, where a self match is 0; DOT_PRODUCT scores as a similarity and can be negative. The vector attribute is excluded from results unless a projection_expression names it and the index projects it. Create the index with create_table's vector_indexes, which is searchable at once. An index added later through update_table refuses with 'Cannot search backfilling vector index' until it settles, for longer than describe_table reports it CREATING, so retry the search rather than waiting for the status."
1528 )]
1529 fn search_vectors(
1530 &self,
1531 Parameters(params): Parameters<SearchVectorsParams>,
1532 ) -> Result<CallToolResult, McpError> {
1533 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1534 return Ok(err);
1535 }
1536 let expression_attribute_values = match parse_optional_dynamo_map(
1537 params.expression_attribute_values,
1538 "expression_attribute_values",
1539 crate::validation::strip_request_validation_tag,
1540 ) {
1541 Ok(v) => v,
1542 Err(err) => return Ok(err),
1543 };
1544
1545 let mut wire = serde_json::json!({
1553 "TableName": params.table_name,
1554 "IndexName": params.index_name,
1555 "SearchVector": params.search_vector,
1556 "TopK": params.top_k,
1557 });
1558 let optional: [(&str, Option<serde_json::Value>); 5] = [
1559 (
1560 "SearchConditionExpression",
1561 params
1562 .search_condition_expression
1563 .map(serde_json::Value::from),
1564 ),
1565 (
1566 "ProjectionExpression",
1567 params.projection_expression.map(serde_json::Value::from),
1568 ),
1569 (
1570 "ExpressionAttributeNames",
1571 params
1572 .expression_attribute_names
1573 .and_then(|v| serde_json::to_value(v).ok()),
1574 ),
1575 (
1576 "ExpressionAttributeValues",
1577 expression_attribute_values.and_then(|v| serde_json::to_value(v).ok()),
1578 ),
1579 (
1580 "ReturnConsumedCapacity",
1581 params.return_consumed_capacity.map(serde_json::Value::from),
1582 ),
1583 ];
1584 for (name, value) in optional {
1585 if let Some(value) = value {
1586 wire[name] = value;
1587 }
1588 }
1589 let request: crate::actions::search_vectors::SearchVectorsRequest =
1590 match parse_json_param(wire, "search_request") {
1591 Ok(r) => r,
1592 Err(e) => return Ok(e),
1593 };
1594 match self.db.search_vectors(request) {
1595 Ok(resp) => {
1596 let json = serde_json::to_value(&resp)
1597 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1598 self.json_result_checked(&json)
1599 }
1600 Err(err) => Ok(to_tool_error(err)),
1601 }
1602 }
1603
1604 #[tool(
1605 description = "Query items using a key condition expression. Supports filtering, projection, pagination, and GSI/LSI queries via IndexName. Use select=COUNT to get only Count and ScannedCount without items."
1606 )]
1607 fn query(
1608 &self,
1609 Parameters(params): Parameters<QueryParams>,
1610 ) -> Result<CallToolResult, McpError> {
1611 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1612 return Ok(err);
1613 }
1614 let expression_attribute_values = match parse_optional_dynamo_map(
1615 params.expression_attribute_values,
1616 "expression_attribute_values",
1617 crate::validation::strip_request_validation_tag,
1618 ) {
1619 Ok(v) => v,
1620 Err(err) => return Ok(err),
1621 };
1622 let exclusive_start_key = match parse_optional_dynamo_map(
1623 params.exclusive_start_key,
1624 "exclusive_start_key",
1625 crate::validation::strip_request_validation_tag,
1626 ) {
1627 Ok(v) => v,
1628 Err(err) => return Ok(err),
1629 };
1630
1631 let request = crate::actions::query::QueryRequest {
1632 table_name: params.table_name,
1633 key_condition_expression: Some(params.key_condition_expression),
1634 filter_expression: params.filter_expression,
1635 projection_expression: params.projection_expression,
1636 expression_attribute_names: params.expression_attribute_names,
1637 expression_attribute_values,
1638 scan_index_forward: params.scan_index_forward.unwrap_or(true),
1639 limit: self.apply_item_limit(params.limit),
1640 exclusive_start_key,
1641 index_name: params.index_name,
1642 select: params.select,
1643 return_consumed_capacity: params.return_consumed_capacity,
1644 ..Default::default()
1645 };
1646 match self.db.query(request) {
1647 Ok(resp) => {
1648 let json = serde_json::to_value(&resp)
1649 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1650 self.json_result_checked(&json)
1651 }
1652 Err(err) => Ok(to_tool_error(err)),
1653 }
1654 }
1655
1656 #[tool(
1657 description = "Scan an entire table or GSI/LSI, optionally applying a filter expression. Use query instead when you know the partition key. Use select=COUNT to get only Count and ScannedCount without items. Use Segment and TotalSegments for parallel scan; segment filtering uses FNV-1a hash at the SQLite level."
1658 )]
1659 fn scan(&self, Parameters(params): Parameters<ScanParams>) -> Result<CallToolResult, McpError> {
1660 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1661 return Ok(err);
1662 }
1663 let expression_attribute_values = match parse_optional_dynamo_map(
1664 params.expression_attribute_values,
1665 "expression_attribute_values",
1666 crate::validation::strip_request_validation_tag,
1667 ) {
1668 Ok(v) => v,
1669 Err(err) => return Ok(err),
1670 };
1671 let exclusive_start_key = match parse_optional_dynamo_map(
1672 params.exclusive_start_key,
1673 "exclusive_start_key",
1674 crate::validation::strip_request_validation_tag,
1675 ) {
1676 Ok(v) => v,
1677 Err(err) => return Ok(err),
1678 };
1679
1680 let request = crate::actions::scan::ScanRequest {
1681 table_name: params.table_name,
1682 filter_expression: params.filter_expression,
1683 projection_expression: params.projection_expression,
1684 expression_attribute_names: params.expression_attribute_names,
1685 expression_attribute_values,
1686 limit: self.apply_item_limit(params.limit),
1687 exclusive_start_key,
1688 index_name: params.index_name,
1689 select: params.select,
1690 segment: params.segment,
1691 total_segments: params.total_segments,
1692 return_consumed_capacity: params.return_consumed_capacity,
1693 ..Default::default()
1694 };
1695 match self.db.scan(request) {
1696 Ok(resp) => {
1697 let json = serde_json::to_value(&resp)
1698 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1699 self.json_result_checked(&json)
1700 }
1701 Err(err) => Ok(to_tool_error(err)),
1702 }
1703 }
1704
1705 #[tool(
1706 description = "[WRITE] Delete a single item by its primary key. Returns the deleted item if return_values is set to ALL_OLD. Blocked if DeletionProtectionEnabled is set on the table. Condition expression values are validated for empty strings/sets/numbers."
1707 )]
1708 fn delete_item(
1709 &self,
1710 Parameters(params): Parameters<DeleteItemParams>,
1711 ) -> Result<CallToolResult, McpError> {
1712 if let Some(err) = self.reject_if_read_only("delete_item") {
1713 return Ok(err);
1714 }
1715 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1716 return Ok(err);
1717 }
1718 let key = match parse_dynamo_map(
1719 params.key,
1720 "key",
1721 crate::validation::envelope_request_validation,
1722 ) {
1723 Ok(v) => v,
1724 Err(err) => return Ok(err),
1725 };
1726 let expression_attribute_values = match parse_optional_dynamo_map(
1727 params.expression_attribute_values,
1728 "expression_attribute_values",
1729 crate::validation::envelope_request_validation,
1730 ) {
1731 Ok(v) => v,
1732 Err(err) => return Ok(err),
1733 };
1734
1735 let request = crate::actions::delete_item::DeleteItemRequest {
1736 table_name: params.table_name,
1737 key,
1738 condition_expression: params.condition_expression,
1739 expression_attribute_names: params.expression_attribute_names,
1740 expression_attribute_values,
1741 return_values: params.return_values,
1742 return_values_on_condition_check_failure: params
1743 .return_values_on_condition_check_failure,
1744 return_consumed_capacity: params.return_consumed_capacity,
1745 ..Default::default()
1746 };
1747 match self.db.delete_item(request) {
1748 Ok(resp) => json_result(&resp),
1749 Err(err) => Ok(to_tool_error(err)),
1750 }
1751 }
1752
1753 #[tool(
1754 description = "[WRITE] Write up to 25 put or delete operations across one or more tables in a single call. Accepts both PascalCase (PutRequest/DeleteRequest) and snake_case (put_request/delete_request) keys. Duplicate keys across operations are rejected. ReturnItemCollectionMetrics: SIZE returns partition collection size for tables with LSIs."
1755 )]
1756 fn batch_write_item(
1757 &self,
1758 Parameters(params): Parameters<BatchWriteItemParams>,
1759 ) -> Result<CallToolResult, McpError> {
1760 if let Some(err) = self.reject_if_read_only("batch_write_item") {
1761 return Ok(err);
1762 }
1763 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1764 return Ok(err);
1765 }
1766 let request_items = match parse_batch_write_request_items(params.request_items) {
1769 Ok(v) => v,
1770 Err(e) => return Ok(e),
1771 };
1772
1773 let request = crate::actions::batch_write_item::BatchWriteItemRequest {
1774 request_items,
1775 return_consumed_capacity: params.return_consumed_capacity,
1776 ..Default::default()
1777 };
1778 match self.db.batch_write_item(request) {
1779 Ok(resp) => json_result(&resp),
1780 Err(err) => Ok(to_tool_error(err)),
1781 }
1782 }
1783
1784 #[tool(
1785 description = "Retrieve up to 100 items from one or more tables in a single call. Accepts both PascalCase (Keys) and snake_case (keys) in request_items."
1786 )]
1787 fn batch_get_item(
1788 &self,
1789 Parameters(params): Parameters<BatchGetItemParams>,
1790 ) -> Result<CallToolResult, McpError> {
1791 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
1792 return Ok(err);
1793 }
1794 let request_items = match parse_batch_get_request_items(params.request_items) {
1795 Ok(v) => v,
1796 Err(e) => return Ok(e),
1797 };
1798
1799 let request = crate::actions::batch_get_item::BatchGetItemRequest {
1800 request_items,
1801 return_consumed_capacity: params.return_consumed_capacity,
1802 };
1803 match self.db.batch_get_item(request) {
1804 Ok(resp) => json_result(&resp),
1805 Err(err) => Ok(to_tool_error(err)),
1806 }
1807 }
1808
1809 #[tool(
1810 description = "Execute a PartiQL statement. Supports SELECT (with LIMIT, nested paths), INSERT (with IF NOT EXISTS, parameter placeholders), UPDATE (with SET arithmetic, REMOVE, nested paths), DELETE. UPDATE and DELETE accept a RETURNING clause, surfaced in the response Items: DELETE allows only RETURNING ALL OLD * (returning the deleted item; other variants are rejected), UPDATE allows all four RETURNING <ALL|MODIFIED> <OLD|NEW> * variants. WHERE supports BETWEEN, IN, CONTAINS, IS MISSING, IS NOT MISSING, AND, OR, NOT, parenthesised grouping up to 64 levels deep, and nested paths; AND binds tighter than OR, and a NOT over a group is applied by De Morgan. An index is named in the FROM clause as \"table\".\"index\", which serves the read from that GSI or LSI on SELECT only; a write statement rejects the qualifier. An UPDATE or DELETE whose WHERE condition fails is a ConditionalCheckFailedException; with return_values_on_condition_check_failure set to ALL_OLD the error carries the item the condition was evaluated against. Write statements are blocked in read-only mode."
1811 )]
1812 fn execute_partiql(
1813 &self,
1814 Parameters(params): Parameters<ExecutePartiqlParams>,
1815 ) -> Result<CallToolResult, McpError> {
1816 if self.config.read_only {
1818 let trimmed = params.statement.trim_start().to_uppercase();
1819 if !trimmed.starts_with("SELECT") {
1820 return Ok(self
1821 .reject_if_read_only("execute_partiql")
1822 .expect("read_only is true"));
1823 }
1824 if params.statement.contains(';') {
1826 return Ok(tool_validation_error(
1827 "AccessDeniedException",
1828 "Multi-statement PartiQL is not allowed in read-only mode",
1829 ));
1830 }
1831 }
1832 let parameters: Option<Vec<AttributeValue>> = match params.parameters {
1835 Some(val) => match serde_json::from_value(val) {
1836 Ok(v) => Some(v),
1837 Err(e) => {
1838 let msg = e.to_string();
1844 if let Some(err) = crate::serde_errors::classify_marked_serde_error(&msg) {
1845 return Ok(to_tool_error(
1846 crate::validation::strip_request_validation_tag(err),
1847 ));
1848 }
1849 return Err(McpError::invalid_params(
1850 format!(
1851 "Invalid parameters: {}",
1852 crate::serde_errors::clean_serde_message(&msg)
1853 ),
1854 None,
1855 ));
1856 }
1857 },
1858 None => None,
1859 };
1860
1861 let request = crate::actions::execute_statement::ExecuteStatementRequest {
1862 statement: params.statement,
1863 parameters,
1864 limit: params.limit,
1865 next_token: params.next_token,
1866 consistent_read: params.consistent_read,
1867 return_consumed_capacity: params.return_consumed_capacity,
1868 return_values_on_condition_check_failure: params
1869 .return_values_on_condition_check_failure,
1870 };
1871 match self.db.execute_statement(request) {
1872 Ok(resp) => json_result(&resp),
1873 Err(err) => Ok(to_tool_error(err)),
1874 }
1875 }
1876
1877 #[tool(
1878 description = "Get database-level information: storage mode, path, size, table count, and per-table summaries. When the data was loaded by dynoxide import, also returns the import's notices and whether any exposures were accepted. Use this as your first call to orient yourself."
1879 )]
1880 fn get_database_info(&self) -> Result<CallToolResult, McpError> {
1881 let db_info = self
1882 .db
1883 .database_info()
1884 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1885
1886 let storage_mode = if db_info.path.is_some() {
1887 "persistent"
1888 } else {
1889 "in-memory"
1890 };
1891
1892 let tables: Vec<serde_json::Value> = db_info
1893 .tables
1894 .into_iter()
1895 .map(|t| {
1896 let mut entry = serde_json::json!({
1897 "table_name": t.stats.table_name,
1898 "item_count": t.stats.item_count,
1899 "size_bytes": t.stats.size_bytes,
1900 });
1901
1902 if let Some(meta) = t.metadata
1904 && let Ok(serde_json::Value::Array(ks)) =
1905 serde_json::from_str::<serde_json::Value>(&meta.key_schema)
1906 {
1907 for elem in &ks {
1908 let name = elem.get("AttributeName").and_then(|v| v.as_str());
1909 let key_type = elem.get("KeyType").and_then(|v| v.as_str());
1910 match (name, key_type) {
1911 (Some(n), Some("HASH")) => {
1912 entry["partition_key"] = serde_json::Value::String(n.to_string());
1913 }
1914 (Some(n), Some("RANGE")) => {
1915 entry["sort_key"] = serde_json::Value::String(n.to_string());
1916 }
1917 _ => {}
1918 }
1919 }
1920 }
1921
1922 entry
1923 })
1924 .collect();
1925
1926 let mut info = serde_json::json!({
1927 "storage_mode": storage_mode,
1928 "path": db_info.path,
1929 "size_bytes": db_info.size_bytes,
1930 "table_count": db_info.table_count,
1931 "tables": tables,
1932 "read_only": self.config.read_only,
1933 "max_items": self.config.max_items,
1934 "max_size_bytes": self.config.max_size_bytes,
1935 });
1936
1937 if let Some(ref data_model) = self.config.data_model {
1938 info["data_model"] = serde_json::to_value(data_model)
1939 .map_err(|e| McpError::internal_error(e.to_string(), None))?;
1940 }
1941
1942 if let Some(ref report) = self.config.import {
1943 info["import"] = report.to_json();
1944 }
1945
1946 json_result(&info)
1947 }
1948
1949 #[tool(
1950 description = "[WRITE] Create a snapshot of the current database state. Returns the snapshot name for later restoration via restore_snapshot. A global limit of 20 snapshots is enforced, and when it is reached auto-snapshots are evicted first, then the oldest manual ones. This holds the database lock for the duration of the copy, so a large database may briefly block other operations."
1951 )]
1952 fn create_snapshot(
1953 &self,
1954 Parameters(params): Parameters<CreateSnapshotParams>,
1955 ) -> Result<CallToolResult, McpError> {
1956 if let Some(err) = self.reject_if_read_only("create_snapshot") {
1957 return Ok(err);
1958 }
1959 match snapshots::create_snapshot(&self.db, params.name.as_deref()) {
1960 Ok(info) => json_result(serde_json::json!({
1961 "name": info.name,
1962 "size_bytes": info.size_bytes,
1963 "message": "Snapshot created successfully",
1964 })),
1965 Err(err) => Ok(to_tool_error(err)),
1966 }
1967 }
1968
1969 #[tool(
1970 description = "[DESTRUCTIVE] Restore the database from a snapshot. This replaces all current data with the snapshot contents. Use list_snapshots to see available snapshot names."
1971 )]
1972 fn restore_snapshot(
1973 &self,
1974 Parameters(params): Parameters<RestoreSnapshotParams>,
1975 ) -> Result<CallToolResult, McpError> {
1976 if let Some(err) = self.reject_if_read_only("restore_snapshot") {
1977 return Ok(err);
1978 }
1979 match snapshots::restore_snapshot(&self.db, ¶ms.name) {
1980 Ok(()) => json_result(serde_json::json!({
1981 "message": "Snapshot restored successfully",
1982 "restored_from": params.name,
1983 })),
1984 Err(err) => Ok(to_tool_error(err)),
1985 }
1986 }
1987
1988 #[tool(
1989 description = "List available database snapshots with name, size, and creation time. Returns newest first, up to 20 entries."
1990 )]
1991 fn list_snapshots(&self) -> Result<CallToolResult, McpError> {
1992 match snapshots::list_snapshots(&self.db, None) {
1993 Ok(snaps) => {
1994 let count = snaps.len();
1995 json_result(serde_json::json!({
1996 "snapshots": snaps,
1997 "count": count,
1998 }))
1999 }
2000 Err(err) => Ok(to_tool_error(err)),
2001 }
2002 }
2003
2004 #[tool(
2005 description = "[DESTRUCTIVE] Delete a snapshot by name. Use list_snapshots to see available snapshots."
2006 )]
2007 fn delete_snapshot(
2008 &self,
2009 Parameters(params): Parameters<DeleteSnapshotParams>,
2010 ) -> Result<CallToolResult, McpError> {
2011 if let Some(err) = self.reject_if_read_only("delete_snapshot") {
2012 return Ok(err);
2013 }
2014 match snapshots::delete_snapshot(&self.db, ¶ms.name) {
2015 Ok(()) => json_result(serde_json::json!({
2016 "message": "Snapshot deleted successfully",
2017 "deleted": params.name,
2018 })),
2019 Err(err) => Ok(to_tool_error(err)),
2020 }
2021 }
2022
2023 #[tool(
2028 description = "[WRITE] Update a table: add/remove GSIs, change stream settings, switch BillingMode, ProvisionedThroughput or OnDemandThroughput, change TableClass, or toggle DeletionProtectionEnabled."
2029 )]
2030 fn update_table(
2031 &self,
2032 Parameters(params): Parameters<UpdateTableParams>,
2033 ) -> Result<CallToolResult, McpError> {
2034 if let Some(err) = self.reject_if_read_only("update_table") {
2035 return Ok(err);
2036 }
2037 let attribute_definitions = match params
2038 .attribute_definitions
2039 .map(serde_json::from_value)
2040 .transpose()
2041 {
2042 Ok(v) => v,
2043 Err(e) => {
2044 return Ok(tool_validation_error(
2045 "InvalidAttributeDefinitions",
2046 &format!("Invalid attribute_definitions: {e}"),
2047 ));
2048 }
2049 };
2050 let global_secondary_index_updates = match params
2051 .global_secondary_index_updates
2052 .map(serde_json::from_value)
2053 .transpose()
2054 {
2055 Ok(v) => v,
2056 Err(e) => {
2057 return Ok(tool_validation_error(
2058 "InvalidGSIUpdates",
2059 &format!("Invalid global_secondary_index_updates: {e}"),
2060 ));
2061 }
2062 };
2063 let canonical_updates = params
2069 .vector_index_updates
2070 .as_ref()
2071 .map(canonicalise_vector_index_updates);
2072 if let Some(ref canonical) = canonical_updates
2077 && let Some(msg) =
2078 crate::actions::update_table::vector_index_updates_request_model_error(canonical)
2079 {
2080 return Ok(tool_validation_error("ValidationException", &msg));
2081 }
2082 let vector_index_updates = match canonical_updates
2083 .as_ref()
2084 .map(crate::actions::update_table::parse_vector_index_updates)
2085 .transpose()
2086 {
2087 Ok(v) => v,
2088 Err(e) => {
2089 return Ok(tool_validation_error(
2090 "InvalidVectorIndexUpdates",
2091 &format!("Invalid vector_index_updates: {e}"),
2092 ));
2093 }
2094 };
2095 let stream_specification = match params
2096 .stream_specification
2097 .map(serde_json::from_value)
2098 .transpose()
2099 {
2100 Ok(v) => v,
2101 Err(e) => {
2102 return Ok(tool_validation_error(
2103 "InvalidStreamSpec",
2104 &format!("Invalid stream_specification: {e}"),
2105 ));
2106 }
2107 };
2108
2109 let provisioned_throughput = match params
2114 .provisioned_throughput
2115 .map(|v| {
2116 serde_json::from_value::<crate::types::ProvisionedThroughput>(v)
2117 .and_then(|pt| serde_json::to_value(&pt))
2118 })
2119 .transpose()
2120 {
2121 Ok(v) => v,
2122 Err(e) => {
2123 return Ok(tool_validation_error(
2124 "InvalidProvisionedThroughput",
2125 &format!("Invalid provisioned_throughput: {e}"),
2126 ));
2127 }
2128 };
2129
2130 let on_demand_throughput = match parse_optional_json_param(
2131 params.on_demand_throughput,
2132 "InvalidOnDemandThroughput",
2133 "on_demand_throughput",
2134 ) {
2135 Ok(v) => v,
2136 Err(e) => return Ok(e),
2137 };
2138
2139 let request = crate::actions::update_table::UpdateTableRequest {
2140 table_name: params.table_name,
2141 attribute_definitions,
2142 global_secondary_index_updates,
2143 vector_index_updates,
2144 stream_specification,
2145 deletion_protection_enabled: params.deletion_protection_enabled,
2146 billing_mode: params.billing_mode,
2147 provisioned_throughput,
2148 table_class: params.table_class,
2149 on_demand_throughput,
2150 };
2151 match self.db.update_table(request) {
2152 Ok(resp) => json_result(&resp),
2153 Err(err) => Ok(to_tool_error(err)),
2154 }
2155 }
2156
2157 #[tool(
2162 description = "[WRITE] Execute up to 100 write actions (Put, Update, Delete, ConditionCheck) atomically. All succeed or all fail. Response includes ItemCollectionMetrics for tables with LSIs."
2163 )]
2164 fn transact_write_items(
2165 &self,
2166 Parameters(params): Parameters<TransactWriteItemsParams>,
2167 ) -> Result<CallToolResult, McpError> {
2168 if let Some(err) = self.reject_if_read_only("transact_write_items") {
2169 return Ok(err);
2170 }
2171 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
2172 return Ok(err);
2173 }
2174 let transact_items = match parse_json_param(params.transact_items, "transact_items") {
2175 Ok(v) => v,
2176 Err(err) => return Ok(err),
2177 };
2178 let request = crate::actions::transact_write_items::TransactWriteItemsRequest {
2179 transact_items,
2180 client_request_token: params.client_request_token,
2181 return_consumed_capacity: params.return_consumed_capacity,
2182 ..Default::default()
2183 };
2184 match self.db.transact_write_items(request) {
2185 Ok(resp) => json_result(&resp),
2186 Err(err) => Ok(to_tool_error(err)),
2187 }
2188 }
2189
2190 #[tool(
2191 description = "Execute up to 100 get operations atomically. Returns a consistent snapshot of all requested items."
2192 )]
2193 fn transact_get_items(
2194 &self,
2195 Parameters(params): Parameters<TransactGetItemsParams>,
2196 ) -> Result<CallToolResult, McpError> {
2197 if let Some(err) = refuse_bad_capacity_mode(params.return_consumed_capacity.as_deref()) {
2198 return Ok(err);
2199 }
2200 let transact_items = match parse_json_param(params.transact_items, "transact_items") {
2201 Ok(v) => v,
2202 Err(err) => return Ok(err),
2203 };
2204 let request = crate::actions::transact_get_items::TransactGetItemsRequest {
2205 transact_items,
2206 return_consumed_capacity: params.return_consumed_capacity,
2207 };
2208 match self.db.transact_get_items(request) {
2209 Ok(resp) => json_result(&resp),
2210 Err(err) => Ok(to_tool_error(err)),
2211 }
2212 }
2213
2214 #[tool(
2219 description = "[WRITE] Execute a batch of PartiQL statements. Each statement can be SELECT, INSERT, UPDATE, or DELETE. A member's RETURNING clause is honoured, with the returned item surfaced on that response's Item field. Write statements are blocked in read-only mode."
2220 )]
2221 fn batch_execute_partiql(
2222 &self,
2223 Parameters(params): Parameters<BatchExecutePartiqlParams>,
2224 ) -> Result<CallToolResult, McpError> {
2225 let statements: Vec<crate::actions::batch_execute_statement::BatchStatementRequest> =
2226 match parse_json_param(params.statements, "statements") {
2227 Ok(v) => v,
2228 Err(err) => return Ok(err),
2229 };
2230
2231 if self.config.read_only {
2233 for stmt in &statements {
2234 let trimmed = stmt.statement.trim_start().to_uppercase();
2235 if !trimmed.starts_with("SELECT") {
2236 return Ok(self
2237 .reject_if_read_only("batch_execute_partiql")
2238 .expect("read_only is true"));
2239 }
2240 if stmt.statement.contains(';') {
2241 return Ok(tool_validation_error(
2242 "AccessDeniedException",
2243 "Multi-statement PartiQL is not allowed in read-only mode",
2244 ));
2245 }
2246 }
2247 }
2248
2249 let request = crate::actions::batch_execute_statement::BatchExecuteStatementRequest {
2252 statements,
2253 return_consumed_capacity: params.return_consumed_capacity,
2254 };
2255 match self.db.batch_execute_statement(request) {
2256 Ok(resp) => json_result(&resp),
2257 Err(err) => Ok(to_tool_error(err)),
2258 }
2259 }
2260
2261 #[tool(
2266 description = "[WRITE] [DESTRUCTIVE] Execute PartiQL statements transactionally (all-or-nothing). Max 100 statements. If any fails, the entire transaction is rolled back. A RETURNING clause is not supported inside a transaction and is rejected with a ValidationException. Example transact_statements: [{\"Statement\": \"INSERT INTO Users VALUE {'pk': 'u1', 'name': 'Alice'}\"}]"
2267 )]
2268 fn execute_transaction_partiql(
2269 &self,
2270 Parameters(params): Parameters<ExecuteTransactionPartiqlParams>,
2271 ) -> Result<CallToolResult, McpError> {
2272 if let Some(err) = self.reject_if_read_only("execute_transaction_partiql") {
2273 return Ok(err);
2274 }
2275 let transact_statements: Vec<crate::actions::execute_transaction::ParameterizedStatement> =
2279 match parse_json_param(params.transact_statements, "transact_statements") {
2280 Ok(v) => v,
2281 Err(err) => return Ok(err),
2282 };
2283
2284 let request = crate::actions::execute_transaction::ExecuteTransactionRequest {
2285 transact_statements,
2286 client_request_token: params.client_request_token,
2287 return_consumed_capacity: params.return_consumed_capacity,
2288 };
2289 match self.db.execute_transaction(request) {
2290 Ok(resp) => json_result(&resp),
2291 Err(err) => Ok(to_tool_error(err)),
2292 }
2293 }
2294
2295 #[tool(
2300 description = "[WRITE] Enable or disable Time to Live (TTL) on a table. Specify the attribute that holds the expiration timestamp."
2301 )]
2302 fn update_time_to_live(
2303 &self,
2304 Parameters(params): Parameters<UpdateTimeToLiveParams>,
2305 ) -> Result<CallToolResult, McpError> {
2306 if let Some(err) = self.reject_if_read_only("update_time_to_live") {
2307 return Ok(err);
2308 }
2309 let time_to_live_specification = match parse_json_param(
2310 params.time_to_live_specification,
2311 "time_to_live_specification",
2312 ) {
2313 Ok(v) => v,
2314 Err(err) => return Ok(err),
2315 };
2316 let request = crate::actions::update_time_to_live::UpdateTimeToLiveRequest {
2317 table_name: params.table_name,
2318 time_to_live_specification,
2319 };
2320 match self.db.update_time_to_live(request) {
2321 Ok(resp) => json_result(&resp),
2322 Err(err) => Ok(to_tool_error(err)),
2323 }
2324 }
2325
2326 #[tool(description = "Get the TTL configuration for a table.")]
2327 fn describe_time_to_live(
2328 &self,
2329 Parameters(params): Parameters<DescribeTimeToLiveParams>,
2330 ) -> Result<CallToolResult, McpError> {
2331 let request = crate::actions::describe_time_to_live::DescribeTimeToLiveRequest {
2332 table_name: params.table_name,
2333 };
2334 match self.db.describe_time_to_live(request) {
2335 Ok(resp) => json_result(&resp),
2336 Err(err) => Ok(to_tool_error(err)),
2337 }
2338 }
2339
2340 #[tool(
2341 description = "[WRITE] Run a TTL sweep, deleting expired items from all TTL-enabled tables. Returns the number of items deleted."
2342 )]
2343 fn sweep_ttl(&self) -> Result<CallToolResult, McpError> {
2344 if let Some(err) = self.reject_if_read_only("sweep_ttl") {
2345 return Ok(err);
2346 }
2347 match self.db.sweep_ttl() {
2348 Ok(count) => json_result(serde_json::json!({
2349 "items_deleted": count,
2350 })),
2351 Err(err) => Ok(to_tool_error(err)),
2352 }
2353 }
2354
2355 #[tool(description = "[WRITE] Add tags to a DynamoDB table.")]
2360 fn tag_resource(
2361 &self,
2362 Parameters(params): Parameters<TagResourceParams>,
2363 ) -> Result<CallToolResult, McpError> {
2364 if let Some(err) = self.reject_if_read_only("tag_resource") {
2365 return Ok(err);
2366 }
2367 let tags = match parse_json_param(params.tags, "tags") {
2368 Ok(v) => v,
2369 Err(err) => return Ok(err),
2370 };
2371 let request = crate::actions::tag_resource::TagResourceRequest {
2372 resource_arn: Some(params.resource_arn),
2373 tags,
2374 };
2375 match self.db.tag_resource(request) {
2376 Ok(resp) => json_result(&resp),
2377 Err(err) => Ok(to_tool_error(err)),
2378 }
2379 }
2380
2381 #[tool(description = "[WRITE] Remove tags from a DynamoDB table.")]
2382 fn untag_resource(
2383 &self,
2384 Parameters(params): Parameters<UntagResourceParams>,
2385 ) -> Result<CallToolResult, McpError> {
2386 if let Some(err) = self.reject_if_read_only("untag_resource") {
2387 return Ok(err);
2388 }
2389 let request = crate::actions::untag_resource::UntagResourceRequest {
2390 resource_arn: Some(params.resource_arn),
2391 tag_keys: params.tag_keys,
2392 };
2393 match self.db.untag_resource(request) {
2394 Ok(resp) => json_result(&resp),
2395 Err(err) => Ok(to_tool_error(err)),
2396 }
2397 }
2398
2399 #[tool(description = "List tags for a DynamoDB table.")]
2400 fn list_tags_of_resource(
2401 &self,
2402 Parameters(params): Parameters<ListTagsOfResourceParams>,
2403 ) -> Result<CallToolResult, McpError> {
2404 let request = crate::actions::list_tags_of_resource::ListTagsOfResourceRequest {
2405 resource_arn: Some(params.resource_arn),
2406 };
2407 match self.db.list_tags_of_resource(request) {
2408 Ok(resp) => json_result(&resp),
2409 Err(err) => Ok(to_tool_error(err)),
2410 }
2411 }
2412
2413 #[tool(description = "List DynamoDB Streams, optionally filtered by table name.")]
2418 fn list_streams(
2419 &self,
2420 Parameters(params): Parameters<ListStreamsParams>,
2421 ) -> Result<CallToolResult, McpError> {
2422 let request = crate::actions::list_streams::ListStreamsRequest {
2423 table_name: params.table_name,
2424 limit: params.limit,
2425 ..Default::default()
2426 };
2427 match self.db.list_streams(request) {
2428 Ok(resp) => json_result(&resp),
2429 Err(err) => Ok(to_tool_error(err)),
2430 }
2431 }
2432
2433 #[tool(description = "Get details about a DynamoDB Stream including its shards.")]
2434 fn describe_stream(
2435 &self,
2436 Parameters(params): Parameters<DescribeStreamParams>,
2437 ) -> Result<CallToolResult, McpError> {
2438 let request = crate::actions::describe_stream::DescribeStreamRequest {
2439 stream_arn: params.stream_arn,
2440 limit: params.limit,
2441 ..Default::default()
2442 };
2443 match self.db.describe_stream(request) {
2444 Ok(resp) => json_result(&resp),
2445 Err(err) => Ok(to_tool_error(err)),
2446 }
2447 }
2448
2449 #[tool(
2450 description = "Get a shard iterator for reading stream records. Use TRIM_HORIZON for all records or LATEST for new records only."
2451 )]
2452 fn get_shard_iterator(
2453 &self,
2454 Parameters(params): Parameters<GetShardIteratorParams>,
2455 ) -> Result<CallToolResult, McpError> {
2456 let request = crate::actions::get_shard_iterator::GetShardIteratorRequest {
2457 stream_arn: params.stream_arn,
2458 shard_id: params.shard_id,
2459 shard_iterator_type: params.shard_iterator_type,
2460 sequence_number: params.sequence_number,
2461 };
2462 match self.db.get_shard_iterator(request) {
2463 Ok(resp) => json_result(&resp),
2464 Err(err) => Ok(to_tool_error(err)),
2465 }
2466 }
2467
2468 #[tool(description = "Read stream records using a shard iterator.")]
2469 fn get_records(
2470 &self,
2471 Parameters(params): Parameters<GetRecordsParams>,
2472 ) -> Result<CallToolResult, McpError> {
2473 let request = crate::actions::get_records::GetRecordsRequest {
2474 shard_iterator: params.shard_iterator,
2475 limit: params.limit,
2476 };
2477 match self.db.get_records(request) {
2478 Ok(resp) => json_result(&resp),
2479 Err(err) => Ok(to_tool_error(err)),
2480 }
2481 }
2482
2483 #[tool(
2484 description = "[WRITE] Bulk-insert items into a table. Faster than repeated put_item calls. Maximum 10,000 items per call. Existing items with the same key are overwritten."
2485 )]
2486 fn bulk_put_items(
2487 &self,
2488 Parameters(params): Parameters<BulkPutItemsParams>,
2489 ) -> Result<CallToolResult, McpError> {
2490 if let Some(err) = self.reject_if_read_only("bulk_put_items") {
2491 return Ok(err);
2492 }
2493
2494 const MAX_ITEMS: usize = 10_000;
2495 if params.items.len() > MAX_ITEMS {
2496 return Ok(tool_validation_error(
2497 "ValidationException",
2498 &format!(
2499 "Too many items: {} exceeds maximum of {MAX_ITEMS} per call",
2500 params.items.len()
2501 ),
2502 ));
2503 }
2504
2505 let mut parsed_items = Vec::with_capacity(params.items.len());
2506 for (i, item_val) in params.items.into_iter().enumerate() {
2507 match parse_dynamo_map(
2508 item_val,
2509 &format!("items[{i}]"),
2510 crate::validation::strip_request_validation_tag,
2511 ) {
2512 Ok(item) => parsed_items.push(item),
2513 Err(err) => return Ok(err),
2514 }
2515 }
2516
2517 let options = crate::ImportOptions {
2518 record_streams: params.record_streams,
2519 ..Default::default()
2520 };
2521
2522 match self
2523 .db
2524 .import_items(¶ms.table_name, parsed_items, options)
2525 {
2526 Ok(result) => json_result(serde_json::json!({
2527 "items_imported": result.items_imported,
2528 "bytes_imported": result.bytes_imported,
2529 })),
2530 Err(err) => Ok(to_tool_error(err)),
2531 }
2532 }
2533}
2534
2535#[tool_handler]
2540impl ServerHandler for McpServer {
2541 fn get_info(&self) -> ServerInfo {
2542 let base_instructions = "\
2543 Dynoxide is a lightweight, embeddable DynamoDB emulator backed by SQLite. \
2544 All data is local: no AWS credentials required.\n\n\
2545 ## Getting started\n\
2546 1. Call `get_database_info` first to see tables, key schemas, and server config.\n\
2547 2. Use `describe_table` for detailed schema of a specific table.\n\n\
2548 ## Key concepts\n\
2549 - Items use DynamoDB JSON format: `{\"pk\": {\"S\": \"value\"}, \"age\": {\"N\": \"30\"}}`\n\
2550 - Tool names prefixed with `[WRITE]` modify data; `[DESTRUCTIVE]` may cause data loss.\n\
2551 - In read-only mode (`--read-only`), all write tools return an error.\n\
2552 - Batch tools accept both PascalCase and snake_case keys (e.g. `PutRequest` or `put_request`).\n\
2553 - The item, query, scan, search_vectors, batch_get_item, batch_write_item, \
2554 transaction and PartiQL tools take `return_consumed_capacity`: TOTAL or INDEXES adds \
2555 the block DynamoDB would return under a `ConsumedCapacity` key, and NONE, the default, \
2556 adds nothing.\n\n\
2557 ## Available tools\n\
2558 Tables: list_tables, describe_table, create_table, delete_table, update_table\n\
2559 Items: put_item, get_item, update_item, delete_item\n\
2560 Batch: batch_write_item, batch_get_item, bulk_put_items\n\
2561 Query: query, scan\n\
2562 Vector search: search_vectors\n\
2563 Transactions: transact_write_items, transact_get_items\n\
2564 PartiQL: execute_partiql, batch_execute_partiql, execute_transaction_partiql\n\
2565 TTL: update_time_to_live, describe_time_to_live, sweep_ttl\n\
2566 Tags: tag_resource, untag_resource, list_tags_of_resource\n\
2567 Streams: list_streams, describe_stream, get_shard_iterator, get_records\n\
2568 Snapshots: create_snapshot, restore_snapshot, list_snapshots, delete_snapshot\n\
2569 Info: get_database_info";
2570
2571 let mut instructions = base_instructions.to_string();
2572 if let Some(report) = &self.config.import {
2573 instructions.push_str("\n\n");
2574 instructions.push_str(&report.instructions_sentence());
2575 }
2576 if let Some(summary) = self
2577 .config
2578 .data_model
2579 .as_ref()
2580 .and_then(|dm| dm.instructions_summary(self.config.data_model_summary_limit))
2581 {
2582 instructions.push_str("\n\n");
2583 instructions.push_str(&summary);
2584 }
2585
2586 ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
2587 .with_server_info(Implementation::new("dynoxide", crate::PRODUCT_VERSION))
2588 .with_instructions(instructions)
2589 }
2590}
2591
2592fn invalid_request_items_error(message: &str) -> CallToolResult {
2599 tool_validation_error("InvalidRequestItems", message)
2600}
2601
2602fn batch_decode_error(context: &str, msg: &str) -> CallToolResult {
2608 match crate::serde_errors::classify_marked_serde_error(msg) {
2609 Some(err) => to_tool_error(crate::validation::strip_request_validation_tag(err)),
2610 None => invalid_request_items_error(&format!(
2611 "{context}: {}",
2612 crate::serde_errors::clean_serde_message(msg)
2613 )),
2614 }
2615}
2616
2617fn parse_batch_write_request_items(
2620 val: serde_json::Value,
2621) -> Result<HashMap<String, Vec<crate::actions::batch_write_item::WriteRequest>>, CallToolResult> {
2622 let map = val
2623 .as_object()
2624 .ok_or_else(|| invalid_request_items_error("request_items must be an object"))?;
2625
2626 let mut result = HashMap::new();
2627 for (table_name, requests) in map {
2628 let arr = requests.as_array().ok_or_else(|| {
2629 invalid_request_items_error(&format!("request_items['{table_name}'] must be an array"))
2630 })?;
2631
2632 let mut write_requests = Vec::new();
2633 for req in arr {
2634 let put = req.get("PutRequest").or_else(|| req.get("put_request"));
2635 let del = req
2636 .get("DeleteRequest")
2637 .or_else(|| req.get("delete_request"));
2638
2639 let put_request = match put {
2640 Some(p) => {
2641 let item_val = p
2642 .get("Item")
2643 .or_else(|| p.get("item"))
2644 .ok_or_else(|| invalid_request_items_error("PutRequest missing Item"))?;
2645 let item: HashMap<String, AttributeValue> =
2646 serde_json::from_value(item_val.clone()).map_err(|e| {
2647 batch_decode_error("Invalid item in PutRequest", &e.to_string())
2648 })?;
2649 Some(crate::actions::batch_write_item::PutRequest { item })
2650 }
2651 None => None,
2652 };
2653
2654 let delete_request = match del {
2655 Some(d) => {
2656 let key_val = d
2657 .get("Key")
2658 .or_else(|| d.get("key"))
2659 .ok_or_else(|| invalid_request_items_error("DeleteRequest missing Key"))?;
2660 let key: HashMap<String, AttributeValue> =
2661 serde_json::from_value(key_val.clone()).map_err(|e| {
2662 batch_decode_error("Invalid key in DeleteRequest", &e.to_string())
2663 })?;
2664 Some(crate::actions::batch_write_item::DeleteRequest { key })
2665 }
2666 None => None,
2667 };
2668
2669 write_requests.push(crate::actions::batch_write_item::WriteRequest {
2670 put_request,
2671 delete_request,
2672 });
2673 }
2674 result.insert(table_name.clone(), write_requests);
2675 }
2676 Ok(result)
2677}
2678
2679fn parse_batch_get_request_items(
2682 val: serde_json::Value,
2683) -> Result<HashMap<String, crate::actions::batch_get_item::KeysAndAttributes>, CallToolResult> {
2684 let map = val
2685 .as_object()
2686 .ok_or_else(|| invalid_request_items_error("request_items must be an object"))?;
2687
2688 let mut result = HashMap::new();
2689 for (table_name, attrs) in map {
2690 let keys_val = attrs
2691 .get("Keys")
2692 .or_else(|| attrs.get("keys"))
2693 .ok_or_else(|| {
2694 invalid_request_items_error(&format!("request_items['{table_name}'] missing Keys"))
2695 })?;
2696
2697 let keys: Vec<HashMap<String, AttributeValue>> = serde_json::from_value(keys_val.clone())
2698 .map_err(|e| {
2699 batch_decode_error(&format!("Invalid keys for '{table_name}'"), &e.to_string())
2700 })?;
2701
2702 let projection_expression = attrs
2703 .get("ProjectionExpression")
2704 .or_else(|| attrs.get("projection_expression"))
2705 .and_then(|v| v.as_str())
2706 .map(String::from);
2707
2708 let expression_attribute_names = attrs
2709 .get("ExpressionAttributeNames")
2710 .or_else(|| attrs.get("expression_attribute_names"))
2711 .and_then(|v| serde_json::from_value(v.clone()).ok());
2712
2713 let consistent_read = attrs
2714 .get("ConsistentRead")
2715 .or_else(|| attrs.get("consistent_read"))
2716 .and_then(|v| v.as_bool());
2717
2718 result.insert(
2719 table_name.clone(),
2720 crate::actions::batch_get_item::KeysAndAttributes {
2721 keys,
2722 projection_expression,
2723 expression_attribute_names,
2724 consistent_read,
2725 attributes_to_get: None,
2726 },
2727 );
2728 }
2729 Ok(result)
2730}
2731
2732fn flatten_table_description(desc: &crate::actions::TableDescription) -> serde_json::Value {
2737 let mut pk = serde_json::Value::Null;
2738 let mut sk = serde_json::Value::Null;
2739
2740 for key in &desc.key_schema {
2741 let key_info = serde_json::json!({
2742 "name": key.attribute_name,
2743 "type": find_attribute_type(&key.attribute_name, &desc.attribute_definitions),
2744 });
2745 if key.key_type == crate::types::KeyType::HASH {
2746 pk = key_info;
2747 } else {
2748 sk = key_info;
2749 }
2750 }
2751
2752 let gsis: Vec<serde_json::Value> = desc
2753 .global_secondary_indexes
2754 .as_ref()
2755 .map(|gsis| {
2756 gsis.iter()
2757 .map(|gsi| {
2758 let mut gsi_pk = serde_json::Value::Null;
2759 let mut gsi_sk = serde_json::Value::Null;
2760 for key in &gsi.key_schema {
2761 let key_info = serde_json::json!({
2762 "name": key.attribute_name,
2763 "type": find_attribute_type(&key.attribute_name, &desc.attribute_definitions),
2764 });
2765 if key.key_type == crate::types::KeyType::HASH {
2766 gsi_pk = key_info;
2767 } else {
2768 gsi_sk = key_info;
2769 }
2770 }
2771 let projection_type = &gsi.projection.projection_type;
2772 serde_json::json!({
2773 "index_name": gsi.index_name,
2774 "partition_key": gsi_pk,
2775 "sort_key": gsi_sk,
2776 "projection_type": projection_type,
2777 "item_count": gsi.item_count.unwrap_or(0),
2778 })
2779 })
2780 .collect()
2781 })
2782 .unwrap_or_default();
2783
2784 let vector_indexes: Vec<serde_json::Value> = desc
2788 .vector_indexes
2789 .as_ref()
2790 .map(|vixs| {
2791 vixs.iter()
2792 .map(|vix| {
2793 let schema: Option<Vec<serde_json::Value>> =
2797 vix.search_schema.as_ref().map(|elems| {
2798 elems
2799 .iter()
2800 .map(|e| {
2801 serde_json::json!({
2802 "attribute_name": e.attribute_name,
2803 "search_schema_element_type":
2804 e.search_schema_element_type,
2805 })
2806 })
2807 .collect()
2808 });
2809 let mut out = serde_json::json!({
2810 "index_name": vix.index_name,
2811 "vector_attribute": {
2816 "attribute_name": vix.vector_attribute.attribute_name
2817 },
2818 "dimensions": vix.dimensions,
2819 "distance_function": vix.distance_function,
2820 "projection": {"projection_type": vix.projection.projection_type},
2821 "item_count": vix.item_count.unwrap_or(0),
2822 "index_status": vix.index_status,
2827 });
2828 if let Some(backfilling) = vix.backfilling {
2833 out["backfilling"] = serde_json::Value::Bool(backfilling);
2834 }
2835 if let Some(schema) = schema {
2836 out["search_schema"] = serde_json::Value::Array(schema);
2837 }
2838 out
2839 })
2840 .collect()
2841 })
2842 .unwrap_or_default();
2843
2844 let stream_enabled = desc.latest_stream_arn.is_some();
2845
2846 let billing_mode = desc
2847 .billing_mode_summary
2848 .as_ref()
2849 .map(|s| s.billing_mode.as_str())
2850 .unwrap_or("PROVISIONED");
2851 let table_class = desc
2852 .table_class_summary
2853 .as_ref()
2854 .map(|s| s.table_class.as_str())
2855 .unwrap_or("STANDARD");
2856
2857 let mut flat = serde_json::json!({
2858 "table_name": desc.table_name,
2859 "status": desc.table_status,
2860 "partition_key": pk,
2861 "sort_key": sk,
2862 "item_count": desc.item_count.unwrap_or(0),
2863 "size_bytes": desc.table_size_bytes.unwrap_or(0),
2864 "gsis": gsis,
2865 "vector_indexes": vector_indexes,
2866 "stream_enabled": stream_enabled,
2867 "billing_mode": billing_mode,
2868 "table_class": table_class,
2869 });
2870
2871 if billing_mode == "PROVISIONED"
2874 && let Some(ref pt) = desc.provisioned_throughput
2875 {
2876 flat["provisioned_throughput"] = serde_json::json!({
2877 "read_capacity_units": pt.read_capacity_units,
2878 "write_capacity_units": pt.write_capacity_units,
2879 });
2880 }
2881 if let Some(ref odt) = desc.on_demand_throughput {
2882 flat["on_demand_throughput"] = serde_json::json!({
2883 "max_read_request_units": odt.max_read_request_units,
2884 "max_write_request_units": odt.max_write_request_units,
2885 });
2886 }
2887
2888 flat
2889}
2890
2891fn enrich_with_ttl(flattened: &mut serde_json::Value, db: &Database, table_name: &str) {
2893 if let Ok(Some(meta)) = db.get_table_metadata(table_name) {
2894 if meta.ttl_enabled {
2895 flattened["ttl"] = serde_json::json!({
2896 "enabled": true,
2897 "attribute": meta.ttl_attribute,
2898 });
2899 } else {
2900 flattened["ttl"] = serde_json::json!({ "enabled": false });
2901 }
2902 }
2903}
2904
2905fn find_attribute_type(
2906 name: &str,
2907 attr_defs: &[crate::types::AttributeDefinition],
2908) -> &'static str {
2909 attr_defs
2910 .iter()
2911 .find(|d| d.attribute_name == name)
2912 .map(|d| match d.attribute_type {
2913 crate::types::ScalarAttributeType::S => "S",
2914 crate::types::ScalarAttributeType::N => "N",
2915 crate::types::ScalarAttributeType::B => "B",
2916 })
2917 .unwrap_or("S")
2918}
2919
2920#[cfg(test)]
2921mod tests {
2922 use super::*;
2923
2924 fn report(exposures: usize, cautions: usize, accepted: bool) -> ImportReport {
2925 let mut notices = Vec::new();
2926 for i in 0..exposures {
2927 notices.push(ImportNotice {
2928 concern: ImportConcern::Exposure,
2929 message: format!("exposure {i}"),
2930 });
2931 }
2932 for i in 0..cautions {
2933 notices.push(ImportNotice {
2934 concern: ImportConcern::Caution,
2935 message: format!("caution {i}"),
2936 });
2937 }
2938 ImportReport {
2939 tables: 1,
2940 items: 1,
2941 skipped: 0,
2942 notices,
2943 exposures_accepted: accepted,
2944 }
2945 }
2946
2947 #[test]
2948 fn several_accepted_exposures_read_as_a_plural() {
2949 assert_eq!(
2950 report(2, 1, true).instructions_sentence(),
2951 "This database was populated by dynoxide import; get_database_info reports 3 notices, \
2952 of which 2 are exposures that were accepted with --accept-exposure."
2953 );
2954 }
2955
2956 #[test]
2957 fn an_exposure_nobody_accepted_is_still_named() {
2958 assert_eq!(
2961 report(1, 0, false).instructions_sentence(),
2962 "This database was populated by dynoxide import; get_database_info reports 1 notice, \
2963 of which 1 is an exposure."
2964 );
2965 }
2966
2967 #[test]
2968 fn the_json_counts_agree_with_the_notices() {
2969 let json = report(1, 2, true).to_json();
2970 assert_eq!(json["notice_count"], 3);
2971 assert_eq!(json["exposure_count"], 1);
2972 assert_eq!(json["exposures_accepted"], true);
2973 assert_eq!(json["notices"][0]["concern"], "exposure");
2974 assert_eq!(json["notices"][1]["concern"], "caution");
2975 assert_eq!(json["notices"].as_array().unwrap().len(), 3);
2976 }
2977}
2978
2979#[cfg(test)]
2984mod capacity_tests {
2985 use super::*;
2986 use serde_json::{Value, json};
2987
2988 const BOGUS_MODE: &str = "1 validation error detected: Value 'BOGUS' at \
2991 'returnConsumedCapacity' failed to satisfy constraint: Member must satisfy \
2992 enum value set: [INDEXES, TOTAL, NONE]";
2993
2994 fn wire<T: serde::de::DeserializeOwned>(value: Value) -> T {
2995 serde_json::from_value(value).expect("fixture request deserialises")
2996 }
2997
2998 fn seeded() -> Database {
3002 let db = Database::memory().unwrap();
3003 db.create_table(wire(json!({
3004 "TableName": "Cap",
3005 "BillingMode": "PAY_PER_REQUEST",
3006 "KeySchema": [
3007 {"AttributeName": "pk", "KeyType": "HASH"},
3008 {"AttributeName": "sk", "KeyType": "RANGE"}
3009 ],
3010 "AttributeDefinitions": [
3011 {"AttributeName": "pk", "AttributeType": "S"},
3012 {"AttributeName": "sk", "AttributeType": "S"},
3013 {"AttributeName": "g", "AttributeType": "S"},
3014 {"AttributeName": "l", "AttributeType": "S"}
3015 ],
3016 "GlobalSecondaryIndexes": [{
3017 "IndexName": "byG",
3018 "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}],
3019 "Projection": {"ProjectionType": "ALL"}
3020 }],
3021 "LocalSecondaryIndexes": [{
3022 "IndexName": "byL",
3023 "KeySchema": [
3024 {"AttributeName": "pk", "KeyType": "HASH"},
3025 {"AttributeName": "l", "KeyType": "RANGE"}
3026 ],
3027 "Projection": {"ProjectionType": "ALL"}
3028 }]
3029 })))
3030 .unwrap();
3031 for (sk, g, l, size) in [
3032 ("s1", Some("g1"), "l1", 600),
3033 ("s2", Some("g1"), "l2", 5000),
3034 ("s3", None, "l3", 9000),
3035 ] {
3036 let mut item = json!({
3037 "pk": {"S": "p"}, "sk": {"S": sk}, "l": {"S": l},
3038 "data": {"S": "x".repeat(size)}
3039 });
3040 if let Some(g) = g {
3041 item["g"] = json!({"S": g});
3042 }
3043 db.put_item(wire(json!({"TableName": "Cap", "Item": item})))
3044 .unwrap();
3045 }
3046 db.create_table(wire(json!({
3047 "TableName": "Vecs",
3048 "BillingMode": "PAY_PER_REQUEST",
3049 "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}],
3050 "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}],
3051 "VectorIndexes": [{
3052 "IndexName": "cosine",
3053 "VectorAttribute": {"AttributeName": "embedding"},
3054 "Dimensions": 3,
3055 "DistanceFunction": "COSINE",
3056 "Projection": {"ProjectionType": "ALL"}
3057 }]
3058 })))
3059 .unwrap();
3060 for (pk, v) in [("a", ["1", "0", "0"]), ("b", ["0", "1", "0"])] {
3061 let embedding: Vec<Value> = v.iter().map(|n| json!({"N": n})).collect();
3062 db.put_item(wire(json!({
3063 "TableName": "Vecs",
3064 "Item": {"pk": {"S": pk}, "embedding": {"L": embedding}}
3065 })))
3066 .unwrap();
3067 }
3068 db
3069 }
3070
3071 fn server() -> McpServer {
3072 McpServer::new(Arc::new(seeded()))
3073 }
3074
3075 fn reply(result: Result<CallToolResult, McpError>) -> (bool, Value) {
3077 let result = result.expect("the tool answers rather than failing the protocol");
3078 let text = &result.content[0]
3079 .as_text()
3080 .expect("tools reply in text")
3081 .text;
3082 (
3083 result.is_error == Some(true),
3084 serde_json::from_str(text).expect("tools reply in JSON"),
3085 )
3086 }
3087
3088 fn by_table(block: Value) -> Value {
3092 match block {
3093 Value::Array(mut entries) => {
3094 entries.sort_by(|a, b| {
3095 a["TableName"]
3096 .as_str()
3097 .unwrap_or_default()
3098 .cmp(b["TableName"].as_str().unwrap_or_default())
3099 });
3100 Value::Array(entries)
3101 }
3102 other => other,
3103 }
3104 }
3105
3106 type ToolCall = fn(&McpServer, Value) -> Result<CallToolResult, McpError>;
3107 type ActionCall = fn(&Database, Value) -> Value;
3108
3109 struct Case {
3112 tool: ToolCall,
3113 args: Value,
3114 action: ActionCall,
3115 request: Value,
3116 }
3117
3118 impl Case {
3119 fn call(&self, server: &McpServer, mode: Option<&str>) -> (bool, Value) {
3120 match mode {
3121 Some(mode) => self.call_with(server, json!(mode)),
3122 None => reply((self.tool)(server, self.args.clone())),
3123 }
3124 }
3125
3126 fn call_with(&self, server: &McpServer, mode: Value) -> (bool, Value) {
3128 let mut args = self.args.clone();
3129 args["return_consumed_capacity"] = mode;
3130 reply((self.tool)(server, args))
3131 }
3132
3133 fn expected(&self, mode: &str) -> Value {
3135 let mut request = self.request.clone();
3136 request["ReturnConsumedCapacity"] = json!(mode);
3137 let body = (self.action)(&seeded(), request);
3138 assert!(
3139 !body["ConsumedCapacity"].is_null(),
3140 "{mode}: the action reported no capacity, so a match would prove nothing"
3141 );
3142 by_table(body["ConsumedCapacity"].clone())
3143 }
3144
3145 fn reported(&self, mode: &str) -> Value {
3147 let (is_error, body) = self.call(&server(), Some(mode));
3148 assert!(!is_error, "{mode}: the tool failed: {body}");
3149 by_table(body["ConsumedCapacity"].clone())
3150 }
3151
3152 fn check(&self) {
3154 for mode in ["TOTAL", "INDEXES"] {
3155 assert_eq!(self.reported(mode), self.expected(mode), "{mode}");
3156 }
3157 for mode in [Some("NONE"), None] {
3158 let (is_error, body) = self.call(&server(), mode);
3159 assert!(!is_error, "{mode:?}: the tool failed: {body}");
3160 assert!(
3161 body.get("ConsumedCapacity").is_none(),
3162 "{mode:?}: unasked capacity in {body}"
3163 );
3164 }
3165 let (is_error, body) = self.call_with(&server(), Value::Null);
3167 assert!(!is_error, "null: the tool failed: {body}");
3168 assert!(
3169 body.get("ConsumedCapacity").is_none(),
3170 "null: unasked capacity in {body}"
3171 );
3172 let (is_error, body) = self.call(&server(), Some("BOGUS"));
3173 assert!(is_error, "BOGUS was accepted: {body}");
3174 assert_eq!(body["error_type"], "ValidationException");
3175 assert_eq!(body["message"], BOGUS_MODE);
3176 }
3177 }
3178
3179 fn get_item_case(consistent: bool) -> Case {
3180 let key = json!({"pk": {"S": "p"}, "sk": {"S": "s2"}});
3181 Case {
3182 tool: |s, v| s.get_item(Parameters(wire(v))),
3183 args: json!({"table_name": "Cap", "key": key, "consistent_read": consistent}),
3184 action: |db, v| serde_json::to_value(db.get_item(wire(v)).unwrap()).unwrap(),
3185 request: json!({"TableName": "Cap", "Key": key, "ConsistentRead": consistent}),
3186 }
3187 }
3188
3189 fn put_item_case() -> Case {
3190 let item = json!({
3191 "pk": {"S": "p"}, "sk": {"S": "s9"}, "g": {"S": "g9"}, "l": {"S": "l9"},
3192 "data": {"S": "y".repeat(1500)}
3193 });
3194 Case {
3195 tool: |s, v| s.put_item(Parameters(wire(v))),
3196 args: json!({"table_name": "Cap", "item": item}),
3197 action: |db, v| serde_json::to_value(db.put_item(wire(v)).unwrap()).unwrap(),
3198 request: json!({"TableName": "Cap", "Item": item}),
3199 }
3200 }
3201
3202 fn query_case(index: Option<&str>) -> Case {
3203 let mut args = json!({
3204 "table_name": "Cap",
3205 "key_condition_expression": "pk = :p",
3206 "expression_attribute_values": {":p": {"S": "p"}}
3207 });
3208 let mut request = json!({
3209 "TableName": "Cap",
3210 "KeyConditionExpression": "pk = :p",
3211 "ExpressionAttributeValues": {":p": {"S": "p"}}
3212 });
3213 if let Some(index) = index {
3214 args["index_name"] = json!(index);
3215 request["IndexName"] = json!(index);
3216 }
3217 Case {
3218 tool: |s, v| s.query(Parameters(wire(v))),
3219 args,
3220 action: |db, v| serde_json::to_value(db.query(wire(v)).unwrap()).unwrap(),
3221 request,
3222 }
3223 }
3224
3225 fn transact_write_case() -> Case {
3226 let items = json!([
3227 {"Put": {"TableName": "Cap", "Item": {
3228 "pk": {"S": "p"}, "sk": {"S": "s8"}, "g": {"S": "g8"}, "l": {"S": "l8"}
3229 }}},
3230 {"ConditionCheck": {
3231 "TableName": "Cap",
3232 "Key": {"pk": {"S": "p"}, "sk": {"S": "s1"}},
3233 "ConditionExpression": "attribute_exists(pk)"
3234 }}
3235 ]);
3236 Case {
3237 tool: |s, v| s.transact_write_items(Parameters(wire(v))),
3238 args: json!({"transact_items": items}),
3239 action: |db, v| {
3240 serde_json::to_value(db.transact_write_items(wire(v)).unwrap()).unwrap()
3241 },
3242 request: json!({"TransactItems": items}),
3243 }
3244 }
3245
3246 #[test]
3247 fn get_item_reports_the_actions_capacity() {
3248 get_item_case(false).check();
3249 }
3250
3251 #[test]
3252 fn a_consistent_get_item_reports_the_doubled_rate() {
3253 let eventual = get_item_case(false).reported("TOTAL");
3254 let consistent = get_item_case(true);
3255 consistent.check();
3256 assert_eq!(
3257 consistent.reported("TOTAL")["CapacityUnits"]
3258 .as_f64()
3259 .unwrap(),
3260 eventual["CapacityUnits"].as_f64().unwrap() * 2.0
3261 );
3262 }
3263
3264 #[test]
3265 fn put_item_reports_the_actions_capacity() {
3266 let case = put_item_case();
3267 case.check();
3268 let indexes = case.reported("INDEXES");
3269 assert!(
3270 indexes["GlobalSecondaryIndexes"]["byG"].is_object(),
3271 "no GSI arm in {indexes}"
3272 );
3273 assert!(
3274 indexes["LocalSecondaryIndexes"]["byL"].is_object(),
3275 "no LSI arm in {indexes}"
3276 );
3277 }
3278
3279 #[test]
3280 fn update_item_reports_the_actions_capacity() {
3281 let key = json!({"pk": {"S": "p"}, "sk": {"S": "s1"}});
3282 Case {
3283 tool: |s, v| s.update_item(Parameters(wire(v))),
3284 args: json!({
3285 "table_name": "Cap", "key": key,
3286 "update_expression": "SET g = :g",
3287 "expression_attribute_values": {":g": {"S": "g3"}}
3288 }),
3289 action: |db, v| serde_json::to_value(db.update_item(wire(v)).unwrap()).unwrap(),
3290 request: json!({
3291 "TableName": "Cap", "Key": key,
3292 "UpdateExpression": "SET g = :g",
3293 "ExpressionAttributeValues": {":g": {"S": "g3"}}
3294 }),
3295 }
3296 .check();
3297 }
3298
3299 #[test]
3300 fn delete_item_reports_the_actions_capacity() {
3301 let key = json!({"pk": {"S": "p"}, "sk": {"S": "s2"}});
3302 Case {
3303 tool: |s, v| s.delete_item(Parameters(wire(v))),
3304 args: json!({"table_name": "Cap", "key": key}),
3305 action: |db, v| serde_json::to_value(db.delete_item(wire(v)).unwrap()).unwrap(),
3306 request: json!({"TableName": "Cap", "Key": key}),
3307 }
3308 .check();
3309 }
3310
3311 #[test]
3312 fn query_reports_the_actions_capacity() {
3313 query_case(None).check();
3314 }
3315
3316 #[test]
3317 fn an_lsi_query_reports_the_actions_capacity() {
3318 let case = query_case(Some("byL"));
3319 case.check();
3320 let indexes = case.reported("INDEXES");
3321 assert!(
3322 indexes["LocalSecondaryIndexes"]["byL"].is_object(),
3323 "no LSI arm in {indexes}"
3324 );
3325 }
3326
3327 #[test]
3328 fn a_count_query_reports_the_actions_capacity() {
3329 let mut case = query_case(None);
3330 case.args["select"] = json!("COUNT");
3331 case.request["Select"] = json!("COUNT");
3332 case.check();
3333 }
3334
3335 #[test]
3336 fn a_capped_query_reports_the_capped_read() {
3337 let config = McpConfig {
3338 max_items: Some(1),
3339 ..McpConfig::default()
3340 };
3341 let capped = McpServer::with_config(Arc::new(seeded()), config);
3342 let case = query_case(None);
3343 let (is_error, body) = case.call(&capped, Some("TOTAL"));
3344 assert!(!is_error, "{body}");
3345 let mut request = case.request.clone();
3346 request["Limit"] = json!(1);
3347 let expected = Case { request, ..case }.expected("TOTAL");
3348 assert_eq!(body["ConsumedCapacity"], expected);
3349 assert_ne!(
3350 body["ConsumedCapacity"],
3351 query_case(None).expected("TOTAL"),
3352 "the cap made no difference to the figure, so this proves nothing"
3353 );
3354 }
3355
3356 #[test]
3357 fn scan_reports_the_actions_capacity() {
3358 Case {
3359 tool: |s, v| s.scan(Parameters(wire(v))),
3360 args: json!({"table_name": "Cap", "filter_expression": "attribute_exists(g)"}),
3361 action: |db, v| serde_json::to_value(db.scan(wire(v)).unwrap()).unwrap(),
3362 request: json!({"TableName": "Cap", "FilterExpression": "attribute_exists(g)"}),
3363 }
3364 .check();
3365 }
3366
3367 #[test]
3368 fn search_vectors_reports_the_actions_capacity() {
3369 let vector = json!([{"N": "1"}, {"N": "0"}, {"N": "0"}]);
3370 Case {
3371 tool: |s, v| s.search_vectors(Parameters(wire(v))),
3372 args: json!({
3373 "table_name": "Vecs", "index_name": "cosine",
3374 "search_vector": vector, "top_k": 1
3375 }),
3376 action: |db, v| serde_json::to_value(db.search_vectors(wire(v)).unwrap()).unwrap(),
3377 request: json!({
3378 "TableName": "Vecs", "IndexName": "cosine",
3379 "SearchVector": vector, "TopK": 1
3380 }),
3381 }
3382 .check();
3383 }
3384
3385 #[test]
3386 fn batch_get_item_reports_the_actions_capacity() {
3387 let items = json!({
3388 "Cap": {"Keys": [
3389 {"pk": {"S": "p"}, "sk": {"S": "s1"}},
3390 {"pk": {"S": "p"}, "sk": {"S": "s3"}}
3391 ]},
3392 "Vecs": {"Keys": [{"pk": {"S": "a"}}]}
3393 });
3394 Case {
3395 tool: |s, v| s.batch_get_item(Parameters(wire(v))),
3396 args: json!({"request_items": items}),
3397 action: |db, v| serde_json::to_value(db.batch_get_item(wire(v)).unwrap()).unwrap(),
3398 request: json!({"RequestItems": items}),
3399 }
3400 .check();
3401 }
3402
3403 #[test]
3404 fn batch_write_item_reports_the_actions_capacity() {
3405 let items = json!({
3406 "Cap": [
3407 {"PutRequest": {"Item": {
3408 "pk": {"S": "p"}, "sk": {"S": "s7"}, "g": {"S": "g7"}, "l": {"S": "l7"}
3409 }}},
3410 {"DeleteRequest": {"Key": {"pk": {"S": "p"}, "sk": {"S": "s1"}}}}
3411 ]
3412 });
3413 Case {
3414 tool: |s, v| s.batch_write_item(Parameters(wire(v))),
3415 args: json!({"request_items": items}),
3416 action: |db, v| serde_json::to_value(db.batch_write_item(wire(v)).unwrap()).unwrap(),
3417 request: json!({"RequestItems": items}),
3418 }
3419 .check();
3420 }
3421
3422 #[test]
3423 fn transact_get_items_reports_the_actions_capacity() {
3424 let items = json!([
3425 {"Get": {"TableName": "Cap", "Key": {"pk": {"S": "p"}, "sk": {"S": "s2"}}}},
3426 {"Get": {"TableName": "Vecs", "Key": {"pk": {"S": "a"}}}}
3427 ]);
3428 Case {
3429 tool: |s, v| s.transact_get_items(Parameters(wire(v))),
3430 args: json!({"transact_items": items}),
3431 action: |db, v| serde_json::to_value(db.transact_get_items(wire(v)).unwrap()).unwrap(),
3432 request: json!({"TransactItems": items}),
3433 }
3434 .check();
3435 }
3436
3437 #[test]
3438 fn transact_write_items_reports_the_actions_capacity() {
3439 transact_write_case().check();
3440 }
3441
3442 #[test]
3443 fn a_replayed_transaction_reports_what_the_action_reports() {
3444 let mut case = transact_write_case();
3445 case.args["client_request_token"] = json!("replay-token");
3446 case.request["ClientRequestToken"] = json!("replay-token");
3447
3448 let server = server();
3449 case.call(&server, Some("TOTAL"));
3450 let (is_error, replayed) = case.call(&server, Some("TOTAL"));
3451 assert!(!is_error, "{replayed}");
3452
3453 let db = seeded();
3454 let mut request = case.request.clone();
3455 request["ReturnConsumedCapacity"] = json!("TOTAL");
3456 (case.action)(&db, request.clone());
3457 let expected = (case.action)(&db, request);
3458 assert!(!expected["ConsumedCapacity"].is_null());
3459 assert_eq!(
3460 by_table(replayed["ConsumedCapacity"].clone()),
3461 by_table(expected["ConsumedCapacity"].clone())
3462 );
3463 }
3464
3465 #[test]
3466 fn execute_partiql_select_reports_the_actions_capacity() {
3467 let statement = "SELECT * FROM \"Cap\" WHERE pk = 'p'";
3468 Case {
3469 tool: |s, v| s.execute_partiql(Parameters(wire(v))),
3470 args: json!({"statement": statement}),
3471 action: |db, v| serde_json::to_value(db.execute_statement(wire(v)).unwrap()).unwrap(),
3472 request: json!({"Statement": statement}),
3473 }
3474 .check();
3475 }
3476
3477 #[test]
3478 fn an_index_qualified_select_reports_the_actions_capacity() {
3479 let statement = "SELECT * FROM \"Cap\".\"byG\" WHERE g = 'g1'";
3480 let case = Case {
3481 tool: |s, v| s.execute_partiql(Parameters(wire(v))),
3482 args: json!({"statement": statement}),
3483 action: |db, v| serde_json::to_value(db.execute_statement(wire(v)).unwrap()).unwrap(),
3484 request: json!({"Statement": statement}),
3485 };
3486 case.check();
3487 let indexes = case.reported("INDEXES");
3488 assert!(
3489 indexes["GlobalSecondaryIndexes"]["byG"].is_object(),
3490 "no GSI arm in {indexes}"
3491 );
3492 }
3493
3494 #[test]
3495 fn execute_partiql_update_reports_the_actions_capacity() {
3496 let statement = "UPDATE \"Cap\" SET g = 'g5' WHERE pk = 'p' AND sk = 's1'";
3497 Case {
3498 tool: |s, v| s.execute_partiql(Parameters(wire(v))),
3499 args: json!({"statement": statement}),
3500 action: |db, v| serde_json::to_value(db.execute_statement(wire(v)).unwrap()).unwrap(),
3501 request: json!({"Statement": statement}),
3502 }
3503 .check();
3504 }
3505
3506 #[test]
3507 fn batch_execute_partiql_reports_the_actions_capacity() {
3508 let statements = json!([
3511 {"Statement": "INSERT INTO \"Cap\" VALUE {'pk': 'p', 'sk': 's1'}"},
3512 {"Statement": "INSERT INTO \"Cap\" VALUE {'pk': 'n', 'sk': 'n1', 'g': 'g6'}"}
3513 ]);
3514 Case {
3515 tool: |s, v| s.batch_execute_partiql(Parameters(wire(v))),
3516 args: json!({"statements": statements}),
3517 action: |db, v| {
3518 serde_json::to_value(db.batch_execute_statement(wire(v)).unwrap()).unwrap()
3519 },
3520 request: json!({"Statements": statements}),
3521 }
3522 .check();
3523 }
3524
3525 #[test]
3526 fn execute_transaction_partiql_reports_the_actions_capacity() {
3527 let statements = json!([
3528 {"Statement": "UPDATE \"Cap\" SET g = 'g4' WHERE pk = 'p' AND sk = 's1'"},
3529 {"Statement": "INSERT INTO \"Cap\" VALUE {'pk': 'n', 'sk': 'n2'}"}
3530 ]);
3531 Case {
3532 tool: |s, v| s.execute_transaction_partiql(Parameters(wire(v))),
3533 args: json!({"transact_statements": statements}),
3534 action: |db, v| serde_json::to_value(db.execute_transaction(wire(v)).unwrap()).unwrap(),
3535 request: json!({"TransactStatements": statements}),
3536 }
3537 .check();
3538 }
3539
3540 #[test]
3545 fn partiql_tools_list_the_mode_with_the_other_member_errors() {
3546 let server = server();
3547 let long = format!("SELECT * FROM \"Cap\" WHERE pk = '{}'", "x".repeat(9_000));
3548 let (is_error, body) = reply(server.execute_partiql(Parameters(wire(json!({
3549 "statement": long,
3550 "return_consumed_capacity": "BOGUS"
3551 })))));
3552 assert!(is_error, "the statement was accepted: {body}");
3553 let message = body["message"].as_str().unwrap_or_default();
3554 assert!(
3555 message.starts_with("2 validation errors detected: Value '"),
3556 "the statement listed {message}"
3557 );
3558 assert!(message.contains("' at 'statement' failed to satisfy constraint"));
3559 assert!(message.ends_with(
3560 "Value 'BOGUS' at 'returnConsumedCapacity' failed to satisfy constraint: \
3561 Member must satisfy enum value set: [INDEXES, TOTAL, NONE]"
3562 ));
3563
3564 let statements: Vec<Value> = (0..26)
3565 .map(|_| json!({"Statement": "SELECT * FROM \"Cap\" WHERE pk = 'p'"}))
3566 .collect();
3567 let (is_error, body) = reply(server.batch_execute_partiql(Parameters(wire(json!({
3568 "statements": statements,
3569 "return_consumed_capacity": "BOGUS"
3570 })))));
3571 assert!(is_error, "the batch was accepted: {body}");
3572 let message = body["message"].as_str().unwrap_or_default();
3573 assert!(
3574 message.starts_with(
3575 "2 validation errors detected: Value 'BOGUS' at 'returnConsumedCapacity'"
3576 ),
3577 "the batch listed {message}"
3578 );
3579 assert!(message.contains("' at 'statements' failed to satisfy constraint"));
3580
3581 let long = format!("SELECT * FROM \"Cap\" WHERE pk = '{}'", "x".repeat(9_000));
3582 let (is_error, body) = reply(server.execute_transaction_partiql(Parameters(wire(json!({
3583 "transact_statements": [{"Statement": long}],
3584 "client_request_token": "t".repeat(40),
3585 "return_consumed_capacity": "BOGUS"
3586 })))));
3587 assert!(is_error, "the transaction was accepted: {body}");
3588 let message = body["message"].as_str().unwrap_or_default();
3589 assert!(
3590 message.starts_with("3 validation errors detected: Value '"),
3591 "the transaction listed {message}"
3592 );
3593 assert!(message.contains("' at 'clientRequestToken'"));
3594 assert!(message.contains("' at 'transactStatements.1.member.statement'"));
3595 assert!(message.ends_with(
3596 "Value 'BOGUS' at 'returnConsumedCapacity' failed to satisfy constraint: \
3597 Member must satisfy enum value set: [INDEXES, TOTAL, NONE]"
3598 ));
3599 }
3600
3601 #[test]
3602 fn a_refused_mode_writes_nothing() {
3603 let server = server();
3604 let key = json!({"pk": {"S": "p"}, "sk": {"S": "w1"}});
3605 let mut item = key.clone();
3606 item["g"] = json!({"S": "gw"});
3607 let attempts: [(ToolCall, Value); 3] = [
3608 (
3609 |s, v| s.put_item(Parameters(wire(v))),
3610 json!({"table_name": "Cap", "item": item}),
3611 ),
3612 (
3613 |s, v| s.batch_write_item(Parameters(wire(v))),
3614 json!({"request_items": {"Cap": [{"PutRequest": {"Item": item}}]}}),
3615 ),
3616 (
3617 |s, v| s.transact_write_items(Parameters(wire(v))),
3618 json!({"transact_items": [{"Put": {"TableName": "Cap", "Item": item}}]}),
3619 ),
3620 ];
3621 for (tool, mut args) in attempts {
3622 args["return_consumed_capacity"] = json!("BOGUS");
3623 let (is_error, body) = reply(tool(&server, args));
3624 assert!(is_error, "BOGUS was accepted: {body}");
3625 assert_eq!(body["message"], BOGUS_MODE);
3626 let (_, got) = reply(server.get_item(Parameters(wire(json!({
3627 "table_name": "Cap", "key": key
3628 })))));
3629 assert!(got.get("Item").is_none(), "a refused request wrote {got}");
3630 }
3631 }
3632
3633 #[test]
3634 fn the_mode_is_case_sensitive() {
3635 let (is_error, body) = get_item_case(false).call(&server(), Some("total"));
3636 assert!(is_error, "lower-case total was accepted: {body}");
3637 assert_eq!(body["message"], BOGUS_MODE.replace("BOGUS", "total"));
3638 }
3639
3640 #[test]
3641 fn a_read_only_server_refuses_a_write_before_checking_the_mode() {
3642 let config = McpConfig {
3643 read_only: true,
3644 ..McpConfig::default()
3645 };
3646 let read_only = McpServer::with_config(Arc::new(seeded()), config);
3647 let (is_error, body) = put_item_case().call(&read_only, Some("BOGUS"));
3648 assert!(is_error);
3649 assert_eq!(body["error_type"], "AccessDeniedException", "{body}");
3650 }
3651
3652 #[test]
3653 fn a_read_only_server_still_reports_select_capacity() {
3654 let config = McpConfig {
3655 read_only: true,
3656 ..McpConfig::default()
3657 };
3658 let read_only = McpServer::with_config(Arc::new(seeded()), config);
3659 let (is_error, body) = reply(read_only.execute_partiql(Parameters(wire(json!({
3660 "statement": "SELECT * FROM \"Cap\" WHERE pk = 'p'",
3661 "return_consumed_capacity": "TOTAL"
3662 })))));
3663 assert!(!is_error, "{body}");
3664 assert!(
3665 body["ConsumedCapacity"].is_object(),
3666 "no capacity in {body}"
3667 );
3668 }
3669
3670 #[test]
3671 fn a_failed_condition_carries_no_capacity() {
3672 let (is_error, body) = reply(server().update_item(Parameters(wire(json!({
3673 "table_name": "Cap",
3674 "key": {"pk": {"S": "p"}, "sk": {"S": "s1"}},
3675 "update_expression": "SET g = :g",
3676 "condition_expression": "attribute_not_exists(pk)",
3677 "expression_attribute_values": {":g": {"S": "g3"}},
3678 "return_consumed_capacity": "TOTAL"
3679 })))));
3680 assert!(is_error, "{body}");
3681 assert_eq!(body["error_type"], "ConditionalCheckFailedException");
3682 assert!(body.get("ConsumedCapacity").is_none(), "{body}");
3683 }
3684}