add_terraform_resource
Store Terraform resource definitions with schema and version details for persistent memory in IaC workflows.
Instructions
Add a new Terraform resource definition with its schema and version information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| provider_id | No | Provider ID | |
| name | Yes | Resource name | |
| resource_type | Yes | Resource type | |
| schema | Yes | Resource schema | |
| version | Yes | Resource version | |
| doc_url | Yes | Documentation URL |
Implementation Reference
- MCP tool handler function that validates arguments, parses JSON schema, calls the database add_terraform_resource function, and returns success or error content.
async def handle_add_terraform_resource(db: Any, arguments: Dict[str, Any], operation_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle add_terraform_resource tool.""" # Validate required arguments required_args = ["provider", "name", "resource_type", "schema", "version", "doc_url"] missing_args = [arg for arg in required_args if arg not in arguments] if missing_args: error_msg = f"Missing required arguments for add_terraform_resource: {', '.join(missing_args)}" logger.error(error_msg, extra={"operation_id": operation_id}) return [types.TextContent( type="text", text=error_msg )] try: # Validate schema is valid JSON try: import json json.loads(arguments["schema"]) except json.JSONDecodeError: error_msg = "Invalid schema format. Schema must be valid JSON." logger.error(error_msg, extra={"operation_id": operation_id}) return [types.TextContent(type="text", text=error_msg)] # Add resource logger.info( "Adding Terraform resource", extra={ "provider": arguments["provider"], "resource_name": arguments["name"], "operation_id": operation_id, }, ) resource_id = add_terraform_resource( db, arguments["provider"], arguments["name"], arguments["resource_type"], arguments["schema"], arguments["version"], arguments["doc_url"], ) return [types.TextContent( type="text", text=f"Added resource {arguments['name']} with ID: {resource_id}" )] except Exception as e: error_msg = f"Failed to add resource: {str(e)}" logger.error(error_msg, extra={"operation_id": operation_id}) return [types.TextContent(type="text", text=error_msg)] - src/iac_memory_mcp_server/tools/terraform.py:311-318 (registration)Registration of the add_terraform_resource handler in the terraform_tool_handlers dictionary used for MCP tool dispatching.
terraform_tool_handlers = { "get_terraform_provider_info": handle_get_terraform_provider_info, "list_provider_resources": handle_list_provider_resources, "get_terraform_resource_info": handle_get_terraform_resource_info, "add_terraform_provider": handle_add_terraform_provider, "add_terraform_resource": handle_add_terraform_resource, "update_provider_version": handle_update_provider_version, } - JSON schema defining the input parameters and validation rules for the add_terraform_resource tool.
"add_terraform_resource": { "type": "object", "description": "Add a new Terraform resource definition with its schema and version information", "required": [ "provider", "name", "resource_type", "schema", "version", "doc_url", ], "properties": { "provider_id": {"type": "string", "description": "Provider ID"}, "name": {"type": "string", "description": "Resource name"}, "resource_type": {"type": "string", "description": "Resource type"}, "schema": {"type": "string", "description": "Resource schema"}, "version": {"type": "string", "description": "Resource version"}, "doc_url": {"type": "string", "description": "Documentation URL"}, }, }, - Core database helper function that performs the actual insertion of the Terraform resource into the database tables terraform_resources and provider_resources, handling transactions and provider lookup.
def add_terraform_resource( db: DatabaseManager, provider_id: str, name: str, resource_type: str, schema: str, version: str, doc_url: str, ) -> str: """Add a new Terraform resource.""" logger.info( "Adding Terraform resource", extra={ "provider_id": provider_id, "resource_name": name, "resource_type": resource_type, "version": version, "operation": "add_terraform_resource", }, ) try: with db.get_connection() as conn: # Look up provider by ID or name provider = conn.execute( """SELECT id FROM terraform_providers WHERE id = ? OR name = ?""", (provider_id, provider_id), ).fetchone() if not provider: raise ValueError(f"Provider '{provider_id}' not found") provider_id = provider["id"] # Ensure we have the numeric ID conn.execute("BEGIN IMMEDIATE") try: # Insert resource cursor = conn.execute( """INSERT INTO terraform_resources (provider_id, name, resource_type, schema, version, doc_url, description, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)""", ( provider_id, name, resource_type, schema, version, doc_url, f"Terraform {resource_type} resource", ), ) resource_id = cursor.lastrowid # Create provider-resource relationship provider_info = conn.execute( "SELECT name FROM terraform_providers WHERE id = ?", (provider_id,) ).fetchone() conn.execute( """INSERT INTO provider_resources (provider_id, resource_id, provider_name, resource_type, schema_version, doc_url, relationship_type, metadata) VALUES (?, ?, ?, ?, ?, ?, 'MANAGED', '{}')""", ( provider_id, resource_id, provider_info["name"], resource_type, version, doc_url, ), ) conn.commit() return str(cursor.lastrowid) except Exception: conn.rollback() raise except sqlite3.Error as e: error_msg = f"Failed to add Terraform resource: {str(e)}" logger.error(error_msg) raise DatabaseError(error_msg)