list_ansible_collections

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/ansible.py:86-130 (handler)

  MCP tool handler that executes the list_ansible_collections tool: fetches collections from DB, applies optional filters, formats output as text.

  async def handle_list_ansible_collections(
      db: Any, arguments: Dict[str, Any], operation_id: str
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle list_ansible_collections tool."""
      try:
          logger.info("Listing Ansible collections", extra={"operation_id": operation_id})
  
          # Get all collections
          collections = list_ansible_collections(db)
  
          # Apply any filters from criteria
          filter_criteria = arguments.get("filter_criteria", {})
          if filter_criteria and "name_pattern" in filter_criteria:
              pattern = re.compile(filter_criteria["name_pattern"])
              collections = [c for c in collections if pattern.match(c["name"])]
  
          # Format output
          if not collections:
              return [TextContent(type="text", text="No collections found")]
  
          output = ["Available Ansible Collections:"]
          for c in collections:
              output.append(
                  f"\n- {c['name']} v{c['version']}"
                  f"\n  Modules: {c['module_count']}"
                  f"\n  Updated: {c['updated_at']}"
                  f"\n  Latest Module Update: {c.get('latest_module_update', 'N/A')}"
                  f"\n  Docs: {c['doc_url']}"
              )
  
          return [TextContent(type="text", text="\n".join(output))]
  
      except Exception as e:
          error_msg = f"Failed to list collections: {str(e)}"
          logger.error(error_msg, extra={"operation_id": operation_id})
          raise McpError(
              types.ErrorData(
                  code=types.INTERNAL_ERROR,
                  message=error_msg,
                  data={
                      "tool": "list_ansible_collections",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:120-136 (schema)

  JSON schema definition for the list_ansible_collections tool parameters.

  "list_ansible_collections": {
      "type": "object",
      "description": "List all cached Ansible collections with basic metadata",
      "required": [],
      "properties": {
          "filter_criteria": {
              "type": "object",
              "description": "Optional filtering criteria",
              "properties": {
                  "name_pattern": {
                      "type": "string",
                      "description": "Regex pattern to filter collection names",
                  }
              },
          }
      },
  },

• src/iac_memory_mcp_server/tools/ansible.py:473-482 (registration)

  Registration of Ansible tool handlers, mapping 'list_ansible_collections' 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,
  }

• src/iac_memory_mcp_server/db/ansible.py:412-449 (helper)

  Database helper function that queries and returns list of Ansible collections with module counts and latest updates.

  def list_ansible_collections(db: DatabaseManager) -> List[Dict]:
      """Get all Ansible collections with basic metadata.
  
      Args:
          db: Database manager instance
  
      Returns:
          List of collections with metadata including module counts
      """
      logger.info(
          "Listing Ansible collections", extra={"operation": "list_ansible_collections"}
      )
  
      try:
          with db.get_connection() as conn:
              conn.execute("PRAGMA busy_timeout = 5000")  # 5 second timeout
  
              # Get collections with module counts
              collections = conn.execute(
                  """
                  SELECT
                      c.*,
                      COUNT(m.id) as module_count,
                      MAX(m.updated_at) as latest_module_update
                  FROM ansible_collections c
                  LEFT JOIN ansible_modules m ON c.id = m.collection_id
                  GROUP BY c.id
                  ORDER BY c.name
                  """
              ).fetchall()
  
              return [dict(c) for c in collections]
  
      except sqlite3.Error as e:
          error_msg = f"Failed to list collections: {str(e)}"
          logger.error(error_msg)
          raise DatabaseError(error_msg)