get_collection_version_history

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/ansible.py:133-174 (handler)

  The main handler function that processes the tool call, extracts arguments, calls the DB helper, formats output as TextContent, and handles errors.

  async def handle_get_collection_version_history(
      db: Any, arguments: Dict[str, Any], operation_id: str
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle get_collection_version_history tool."""
      try:
          logger.info(
              "Getting collection version history",
              extra={
                  "collection_name": arguments["collection_name"],
                  "operation_id": operation_id,
              },
          )
  
          # Get version history
          versions = get_collection_version_history(db, arguments["collection_name"])
  
          # Format output
          output = [f"Version History for {arguments['collection_name']}:"]
          for v in versions:
              output.append(
                  f"\nVersion: {v['version']}"
                  f"\n  Added: {v['created_at']}"
                  f"\n  Last Updated: {v['updated_at']}"
                  f"\n  Source: {v['source_url']}"
                  f"\n  Docs: {v['doc_url']}"
              )
  
          return [TextContent(type="text", text="\n".join(output))]
  
      except Exception as e:
          error_msg = f"Failed to get collection version history: {str(e)}"
          logger.error(error_msg, extra={"operation_id": operation_id})
          raise McpError(
              types.ErrorData(
                  code=types.INTERNAL_ERROR,
                  message=error_msg,
                  data={
                      "tool": "get_collection_version_history",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:137-147 (schema)

  JSON schema defining the input parameters for the tool, requiring 'collection_name'.

  "get_collection_version_history": {
      "type": "object",
      "description": "Retrieve version history for a specific Ansible collection",
      "required": ["collection_name"],
      "properties": {
          "collection_name": {
              "type": "string",
              "description": "Name of the Ansible collection",
          }
      },
  },

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

  Dictionary mapping tool names to their handler functions, registering 'get_collection_version_history' to handle_get_collection_version_history.

  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:451-494 (helper)

  Database helper function that queries the sqlite database for version history of a given Ansible collection and returns list of dicts.

  def get_collection_version_history(
      db: DatabaseManager, collection_name: str
  ) -> List[Dict]:
      """Get version history for a specific Ansible collection.
  
      Args:
          db: Database manager instance
          collection_name: Name of the collection
  
      Returns:
          List of version entries with timestamps and URLs
      """
      logger.info(
          "Getting collection version history",
          extra={
              "collection_name": collection_name,
              "operation": "get_collection_version_history",
          },
      )
  
      try:
          with db.get_connection() as conn:
              conn.execute("PRAGMA busy_timeout = 5000")  # 5 second timeout
  
              versions = conn.execute(
                  """
                  SELECT version, source_url, doc_url, created_at, updated_at
                  FROM ansible_collections
                  WHERE name = ?
                  ORDER BY created_at DESC
                  """,
                  (collection_name,),
              ).fetchall()
  
              if not versions:
                  raise DatabaseError(f"Collection '{collection_name}' not found")
  
              return [dict(v) for v in versions]
  
      except sqlite3.Error as e:
          error_msg = f"Failed to get collection version history: {str(e)}"
          logger.error(error_msg)
          raise DatabaseError(error_msg)