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
- 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)}) - 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"] }, ) ] - 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)}")] - 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