add_ansible_module
Define and store new Ansible modules with schema, version, and documentation details on the IAC Memory MCP Server for streamlined infrastructure-as-code management.
Instructions
Add a new Ansible module definition with its schema and version information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection | Yes | Collection ID or name | |
| doc_url | Yes | Documentation URL | |
| module_type | Yes | Module type | |
| name | Yes | Module name | |
| schema | Yes | Module schema | |
| version | Yes | Module version |
Implementation Reference
- MCP tool handler function that validates arguments, calls the database add_ansible_module function, and returns success/error messages.async def handle_add_ansible_module(db: Any, arguments: Dict[str, Any], operation_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle add_ansible_module tool.""" try: # Validate module type if arguments["module_type"] not in VALID_MODULE_TYPES: error_msg = f"Invalid module type: {arguments['module_type']}. Valid types are: {', '.join(sorted(VALID_MODULE_TYPES))}" logger.error(error_msg, extra={"operation_id": operation_id}) return [TextContent(type="text", text=error_msg)] logger.info( "Adding Ansible module", extra={ "collection": arguments["collection"], "module_name": arguments["name"], "operation_id": operation_id, }, ) # Add module module_id = add_ansible_module( db, arguments["collection"], arguments["name"], arguments["module_type"], arguments["schema"], arguments["version"], arguments["doc_url"], ) return [TextContent( type="text", text=f"Added module {arguments['name']} with ID: {module_id}" )] except Exception as e: error_msg = f"Failed to add module: {str(e)}" logger.error(error_msg, extra={"operation_id": operation_id}) raise McpError( types.ErrorData( code=types.INTERNAL_ERROR, message=error_msg, data={ "tool": "add_ansible_module", "operation_id": operation_id, }, ) )
- src/iac_memory_mcp_server/tools/ansible.py:559-570 (registration)Registration of the add_ansible_module tool handler in the ansible_tool_handlers dictionary.ansible_tool_handlers = { "get_ansible_collection_info": handle_get_ansible_collection_info, "list_ansible_collections": handle_list_ansible_collections, "get_collection_version_history": handle_get_collection_version_history, "get_ansible_module_info": handle_get_ansible_module_info, "list_collection_modules": handle_list_collection_modules, "get_module_version_compatibility": handle_get_module_version_compatibility, "add_ansible_collection": handle_add_ansible_collection, "add_ansible_module": handle_add_ansible_module, "update_collection_version": handle_update_collection_version, "update_module_version": handle_update_module_version, }
- JSON schema definition for the add_ansible_module tool inputs, including required fields and descriptions."add_ansible_module": { "type": "object", "description": "Add a new Ansible module definition with its schema and version information", "required": [ "collection", "name", "module_type", "schema", "version", "doc_url", ], "properties": { "collection": {"type": "string", "description": "Collection ID or name"}, "name": {"type": "string", "description": "Module name"}, "module_type": {"type": "string", "description": "Module type"}, "schema": {"type": "string", "description": "Module schema"}, "version": {"type": "string", "description": "Module version"}, "doc_url": {"type": "string", "description": "Documentation URL"}, }, },
- Core database function that performs the INSERT into ansible_modules table after validating collection exists.def add_ansible_module( db: DatabaseManager, collection_id: str, name: str, module_type: str, schema: str, version: str, doc_url: str, ) -> str: """Add a new Ansible module.""" try: with db.get_connection() as conn: # Set busy timeout before any operations conn.execute( "PRAGMA busy_timeout = 5000" ) # 5 second timeout per testing rules conn.execute("BEGIN IMMEDIATE") # Start transaction try: # Look up collection by ID or name collection = conn.execute( """SELECT id FROM ansible_collections WHERE id = ? OR name = ?""", (collection_id, collection_id), ).fetchone() if not collection: raise DatabaseError( f"Collection '{collection_id}' not found. " f"Operation timed out after 5 seconds." ) collection_id = collection["id"] # Ensure we have the numeric ID cursor = conn.execute( """INSERT INTO ansible_modules (collection_id, name, type, schema, description, version, doc_url) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( collection_id, name, module_type, schema, module_type, # Using type as description for now version, doc_url, ), ) module_id = str(cursor.lastrowid) conn.commit() return module_id except Exception: conn.rollback() raise except sqlite3.Error as e: raise DatabaseError( f"Failed to add Ansible module: {str(e)}. " f"Operation timed out after 5 seconds." )