aws_resources_query_or_modify

Execute Python code with boto3 to query or modify AWS resources directly from conversations using the AWS Resources MCP Server.

Instructions

Execute a boto3 code snippet to query or modify AWS resources

Input Schema

TableJSON Schema

Name	Required	Description	Default
`code_snippet`	Yes	Python code using boto3 to query or modify AWS resources. The code should have default execution setting variable named 'result'. Example code: 'result = boto3.client('s3').list_buckets()'

Implementation Reference

src/mcp_server_aws_resources/server.py:84-154 (handler)

Core handler that executes the boto3 code snippet provided by the tool input, performing safety validations and returning JSON results.

def execute_query(self, code_snippet: str) -> str:
    """
    Execute a boto3 code snippet and return the results

    Args:
        code_snippet (str): Python code using boto3 to query AWS resources

    Returns:
        str: JSON string containing the query results or error message
    """
    try:
        # Parse the code into an AST
        tree = ast.parse(code_snippet)

        # Analyze the code
        executor = CodeExecutor()
        executor.visit(tree)

        # Validate imports
        allowed_modules = {'boto3', 'operator', 'json', 'datetime', 'pytz', 'dateutil', 're', 'time'}
        unauthorized_imports = executor.imported_modules - allowed_modules
        if unauthorized_imports:
            return json.dumps({
                "error": f"Unauthorized imports: {', '.join(unauthorized_imports)}. "
                        f"Only {', '.join(allowed_modules)} are allowed."
            })

        # Create execution namespace
        local_ns = {
            'boto3': boto3,
            'session': self.session,
            'result': None,
            'itemgetter': itemgetter,
            '__builtins__': {
                name: getattr(__builtins__, name)
                for name in [
                    'dict', 'list', 'tuple', 'set', 'str', 'int', 'float', 'bool',
                    'len', 'max', 'min', 'sorted', 'filter', 'map', 'sum', 'any', 'all',
                    '__import__', 'hasattr', 'getattr', 'isinstance', 'print'
                ]
            }
        }

        # Compile and execute the code
        compiled_code = compile(tree, '<string>', 'exec')
        exec(compiled_code, local_ns)

        # Get the result
        result = local_ns.get('result')

        # Validate result was set
        if not executor.has_result:
            return json.dumps({
                "error": "Code must set a 'result' variable with the query output"
            })

        # Convert result to JSON-serializable format
        if result is not None:
            if hasattr(result, 'to_dict'):
                result = result.to_dict()
            return json.dumps(result, default=str)
        else:
            return json.dumps({"error": "Result cannot be None"})

    except SyntaxError as e:
        logger.error(f"Syntax error in code: {str(e)}")
        return json.dumps({"error": f"Syntax error: {str(e)}"})
    except Exception as e:
        logger.error(f"Error executing query: {str(e)}")
        return json.dumps({"error": str(e)})

src/mcp_server_aws_resources/server.py:176-185 (schema)

Input schema definition for the tool, specifying the required 'code_snippet' parameter.

inputSchema={
    "type": "object",
    "properties": {
        "code_snippet": {
            "type": "string",
            "description": "Python code using boto3 to query or modify AWS resources. The code should have default execution setting variable named 'result'. Example code: 'result = boto3.client('s3').list_buckets()'"
        }
    },
    "required": ["code_snippet"]
},

src/mcp_server_aws_resources/server.py:172-187 (registration)

Registration of the tool in the MCP server's list_tools method, including name, description, and schema.

return [
    types.Tool(
        name="aws_resources_query_or_modify",
        description="Execute a boto3 code snippet to query or modify AWS resources",
        inputSchema={
            "type": "object",
            "properties": {
                "code_snippet": {
                    "type": "string",
                    "description": "Python code using boto3 to query or modify AWS resources. The code should have default execution setting variable named 'result'. Example code: 'result = boto3.client('s3').list_buckets()'"
                }
            },
            "required": ["code_snippet"]
        },
    )
]

src/mcp_server_aws_resources/server.py:189-206 (handler)

MCP server call_tool handler that checks the tool name and invokes the execution logic.

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """Handle tool execution requests"""
    try:
        if name == "aws_resources_query_or_modify":
            if not arguments or "code_snippet" not in arguments:
                raise ValueError("Missing code_snippet argument")

            results = aws_querier.execute_query(arguments["code_snippet"])
            return [types.TextContent(type="text", text=str(results))]
        else:
            raise ValueError(f"Unknown tool: {name}")

    except Exception as e:
        return [types.TextContent(type="text", text=f"Error: {str(e)}")]

src/mcp_server_aws_resources/server.py:43-67 (helper)

Helper class for AST analysis to validate code snippets: checks for 'result' assignment and tracks imports.

class CodeExecutor(ast.NodeTransformer):
    """Custom AST NodeTransformer to validate and transform the code"""

    def __init__(self):
        self.has_result = False
        self.imported_modules = set()

    def visit_Assign(self, node):
        """Track if 'result' variable is assigned"""
        for target in node.targets:
            if isinstance(target, ast.Name) and target.id == 'result':
                self.has_result = True
        return node

    def visit_Import(self, node):
        """Track imported modules"""
        for alias in node.names:
            self.imported_modules.add(alias.name)
        return node

    def visit_ImportFrom(self, node):
        """Track imported modules"""
        self.imported_modules.add(node.module)
        return node

AWS Resources MCP Server