add_ansible_collection

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/ansible.py:459-505 (handler)

  MCP tool handler function that validates input arguments, calls the database function to add the Ansible collection, logs the operation, and returns a success message with the collection ID or an error.

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

• src/iac_memory_mcp_server/db/tools.py:234-244 (schema)

  JSON schema definition for the add_ansible_collection tool, specifying required parameters (name, version, source_url, doc_url) and their descriptions.

  "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:559-570 (registration)

  Registration of the add_ansible_collection tool handler in the ansible_tool_handlers dictionary, mapping the tool name 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,
      "update_collection_version": handle_update_collection_version,
      "update_module_version": handle_update_module_version,
  }

• src/iac_memory_mcp_server/db/ansible.py:13-41 (helper)

  Database helper function that performs the SQL INSERT into ansible_collections table to add the new collection and returns the generated ID.

  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."
          )