dynamodb_item_batch_write

Perform batch write operations to put or delete multiple items in a DynamoDB table efficiently.

Instructions

Batch write operations (put/delete) for DynamoDB items

Input Schema

TableJSON Schema

Name	Required	Description
`table_name`	Yes	Name of the DynamoDB table
`operation`	Yes	Type of batch operation (put or delete)
`items`	Yes	Array of items to process
`key_attributes`	No	For delete operations, specify which attributes form the key

Implementation Reference

src/mcp_server_aws/server.py:265-332 (handler)

The core handler logic within handle_dynamodb_operations that performs batch write (put/delete) operations on DynamoDB items, including batching up to 25 items, formatting data types, retrying unprocessed items, and reporting results.

elif name == "dynamodb_item_batch_write":
    table_name = arguments["table_name"]
    operation = arguments["operation"]
    items = arguments["items"]

    if not items:
        raise ValueError("No items provided for batch operation")

    batch_size = 25
    total_items = len(items)
    processed_items = 0
    failed_items = []

    for i in range(0, total_items, batch_size):
        batch = items[i:i + batch_size]
        request_items = {table_name: []}

        for item in batch:
            if operation == "put":
                formatted_item = {k: get_dynamodb_type(
                    v) for k, v in item.items()}
                request_items[table_name].append({
                    'PutRequest': {'Item': formatted_item}
                })
            elif operation == "delete":
                key_attrs = arguments.get(
                    "key_attributes", list(item.keys()))
                formatted_key = {k: get_dynamodb_type(
                    item[k]) for k in key_attrs}
                request_items[table_name].append({
                    'DeleteRequest': {'Key': formatted_key}
                })

        try:
            response = dynamodb_client.batch_write_item(
                RequestItems=request_items)
            processed_items += len(batch) - len(
                response.get('UnprocessedItems', {}
                             ).get(table_name, [])
            )

            unprocessed = response.get('UnprocessedItems', {})
            retry_count = 0
            max_retries = 3
            while unprocessed and retry_count < max_retries:
                await asyncio.sleep(2 ** retry_count)
                retry_response = dynamodb_client.batch_write_item(
                    RequestItems=unprocessed)
                unprocessed = retry_response.get(
                    'UnprocessedItems', {})
                retry_count += 1

            if unprocessed:
                failed_items.extend([
                    item['PutRequest']['Item'] if 'PutRequest' in item else item['DeleteRequest']['Key']
                    for item in unprocessed.get(table_name, [])
                ])

        except Exception as e:
            logger.error(f"Error processing batch: {str(e)}")
            failed_items.extend(batch)

    response = {
        "total_items": total_items,
        "processed_items": processed_items,
        "failed_items": len(failed_items),
        "failed_items_details": failed_items if failed_items else None
    }

src/mcp_server_aws/tools.py:356-385 (schema)

The Tool schema definition including input validation schema for parameters: table_name, operation (put/delete), items, and optional key_attributes.

Tool(
    name="dynamodb_item_batch_write",
    description="Batch write operations (put/delete) for DynamoDB items",
    inputSchema={
        "type": "object",
        "properties": {
            "table_name": {
                "type": "string",
                "description": "Name of the DynamoDB table"
            },
            "operation": {
                "type": "string",
                "enum": ["put", "delete"],
                "description": "Type of batch operation (put or delete)"
            },
            "items": {
                "type": "array",
                "description": "Array of items to process"
            },
            "key_attributes": {
                "type": "array",
                "description": "For delete operations, specify which attributes form the key",
                "items": {
                    "type": "string"
                }
            }
        },
        "required": ["table_name", "operation", "items"]
    }
),

src/mcp_server_aws/server.py:136-140 (registration)
The MCP server registration point where list_tools returns get_aws_tools(), which includes the dynamodb_item_batch_write tool from get_dynamodb_tools().
```
async def list_tools() -> list[Tool]:
    """List available AWS tools"""
    logger.debug("Handling list_tools request")
    return get_aws_tools()
```

src/mcp_server_aws/utils.py:1-17 (helper)

Helper utility to convert Python data types to DynamoDB AttributeValue format, used in formatting items and keys for batch_write_item.

def get_dynamodb_type(value):
    if isinstance(value, str):
        return {'S': value}
    elif isinstance(value, (int, float)):
        return {'N': str(value)}
    elif isinstance(value, bool):
        return {'BOOL': value}
    elif value is None:
        return {'NULL': True}
    elif isinstance(value, list):
        return {'L': [get_dynamodb_type(v) for v in value]}
    elif isinstance(value, dict):
        return {'M': {k: get_dynamodb_type(v) for k, v in value.items()}}
    else:
        raise ValueError(
            f"Unsupported type for DynamoDB: {type(value)}")

AWS MCP Server