add_ansible_collection
Add a new Ansible collection to the memory store with version, source repository, and documentation URLs for persistent IaC resource management and version tracking.
Instructions
Add a new Ansible collection to the memory store with version and documentation information
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| doc_url | Yes | Documentation URL | |
| name | Yes | Collection name | |
| source_url | Yes | Source repository URL | |
| version | Yes | Collection version |
Input Schema (JSON Schema)
{
"description": "Add a new Ansible collection to the memory store with version and documentation information",
"properties": {
"doc_url": {
"description": "Documentation URL",
"type": "string"
},
"name": {
"description": "Collection name",
"type": "string"
},
"source_url": {
"description": "Source repository URL",
"type": "string"
},
"version": {
"description": "Collection version",
"type": "string"
}
},
"required": [
"name",
"version",
"source_url",
"doc_url"
],
"type": "object"
}
Implementation Reference
- MCP tool handler function that validates arguments, calls the DB helper, and returns success/error response.async def handle_add_ansible_collection(db: Any, arguments: Dict[str, Any], operation_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """Handle add_ansible_collection tool.""" # Validate required arguments required_args = ["name", "version", "source_url", "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_ansible_collection: {', '.join(missing_args)}" logger.error(error_msg, extra={"operation_id": operation_id}) return [TextContent(type="text", text=error_msg)] try: logger.info( "Adding Ansible collection", extra={ "collection_name": arguments["name"], "version": arguments["version"], "operation_id": operation_id, }, ) # Add collection collection_id = add_ansible_collection( db, arguments["name"], arguments["version"], arguments["source_url"], arguments["doc_url"], ) return [TextContent( type="text", text=f"Added collection {arguments['name']} with ID: {collection_id}" )] except Exception as e: error_msg = f"Failed to add collection: {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_collection", "operation_id": operation_id, }, ) )
- JSON schema definition for the tool's input parameters and validation."add_ansible_collection": { "type": "object", "description": "Add a new Ansible collection to the memory store with version and documentation information", "required": ["name", "version", "source_url", "doc_url"], "properties": { "name": {"type": "string", "description": "Collection name"}, "version": {"type": "string", "description": "Collection version"}, "source_url": {"type": "string", "description": "Source repository URL"}, "doc_url": {"type": "string", "description": "Documentation URL"}, }, },
- src/iac_memory_mcp_server/tools/ansible.py:473-482 (registration)Registration of the 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, }
- Database helper function that performs the SQL INSERT to add the Ansible collection.def add_ansible_collection( db: DatabaseManager, name: str, version: str, source_url: str, doc_url: str ) -> str: """Add a new Ansible collection.""" 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: cursor = conn.execute( """INSERT INTO ansible_collections (name, version, source_url, doc_url) VALUES (?, ?, ?, ?)""", (name, version, source_url, doc_url), ) collection_id = str(cursor.lastrowid) conn.commit() return collection_id except Exception: conn.rollback() raise except sqlite3.Error as e: raise DatabaseError( f"Failed to add Ansible collection: {str(e)}. Operation timed out after 5 seconds." )