add_ansible_module
Define and register new Ansible modules with schema details, version information, and documentation links for Infrastructure-as-Code management and version tracking.
Instructions
Add a new Ansible module definition with its schema and version information
Input 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 |
Input Schema (JSON Schema)
{
"description": "Add a new Ansible module definition with its schema and version information",
"properties": {
"collection": {
"description": "Collection ID or name",
"type": "string"
},
"doc_url": {
"description": "Documentation URL",
"type": "string"
},
"module_type": {
"description": "Module type",
"type": "string"
},
"name": {
"description": "Module name",
"type": "string"
},
"schema": {
"description": "Module schema",
"type": "string"
},
"version": {
"description": "Module version",
"type": "string"
}
},
"required": [
"collection",
"name",
"module_type",
"schema",
"version",
"doc_url"
],
"type": "object"
}
Implementation Reference
- MCP tool handler function that performs input validation, logging, delegates to DB helper, and formats the response.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, }, ) )
- JSON schema definition for the add_ansible_module tool inputs, used for validation."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"}, }, },
- src/iac_memory_mcp_server/tools/ansible.py:473-482 (registration)Registration of Ansible tool handlers, mapping 'add_ansible_module' to its handler function.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, }
- Core database helper function that performs the actual INSERT into ansible_modules table after resolving collection ID.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." )