IAC Memory MCP Server

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/ansible.py:375-416 (handler)

  MCP tool handler function that processes arguments, calls the database update function, handles errors, and returns success message.

  async def handle_update_collection_version(
      db: Any, arguments: Dict[str, Any], operation_id: str
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle update_collection_version tool."""
      try:
          logger.info(
              "Updating collection version",
              extra={
                  "collection_id": arguments["collection_id"],
                  "new_version": arguments["new_version"],
                  "operation_id": operation_id,
              },
          )
  
          # Update collection version
          result = update_collection_version(
              db,
              arguments["collection_id"],
              arguments["new_version"],
              arguments.get("new_source_url"),
              arguments.get("new_doc_url"),
          )
  
          return [TextContent(
              type="text",
              text=f"Updated collection version: {result}"
          )]
  
      except Exception as e:
          error_msg = f"Failed to update collection version: {str(e)}"
          logger.error(error_msg, extra={"operation_id": operation_id})
          raise McpError(
              types.ErrorData(
                  code=types.INTERNAL_ERROR,
                  message=error_msg,
                  data={
                      "tool": "update_collection_version",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:245-255 (schema)

  JSON schema defining input parameters for the update_collection_version tool, including required fields and descriptions.

  "update_collection_version": {
      "type": "object",
      "description": "Update an existing Ansible collection's version information and documentation links",
      "required": ["collection_id", "new_version"],
      "properties": {
          "collection_id": {"type": "string", "description": "Collection ID"},
          "new_version": {"type": "string", "description": "New version"},
          "new_source_url": {"type": "string", "description": "New source URL"},
          "new_doc_url": {"type": "string", "description": "New documentation URL"},
      },
  },

• src/iac_memory_mcp_server/tools/ansible.py:559-570 (registration)

  Dictionary mapping tool names to their handler functions, registering update_collection_version with its handler.

  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:83-129 (helper)

  Core database function that performs the SQL UPDATE on ansible_collections table to update version and optional URLs, with transaction handling and error management.

  def update_collection_version(
      db: DatabaseManager,
      collection_id: str,
      new_version: str,
      new_source_url: Optional[str] = None,
      new_doc_url: Optional[str] = None,
  ) -> bool:
      """Update an Ansible collection's version and optional URLs."""
      try:
          updates = ["version = ?"]
          params = [new_version]
  
          if new_source_url:
              updates.append("source_url = ?")
              params.append(new_source_url)
          if new_doc_url:
              updates.append("doc_url = ?")
              params.append(new_doc_url)
  
          updates.append("updated_at = CURRENT_TIMESTAMP")
          params.append(collection_id)
  
          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(
                      f"""UPDATE ansible_collections
                      SET {', '.join(updates)}
                      WHERE id = ?""",
                      tuple(params),
                  )
                  success = cursor.rowcount > 0
                  conn.commit()
                  return success
              except Exception:
                  conn.rollback()
                  raise
      except sqlite3.Error as e:
          raise DatabaseError(
              f"Failed to update collection version: {str(e)}. "
              f"Operation timed out after 5 seconds."
          )