Event Log Analysis in mcp-airflow-api: Query, Retrieve, and Aggregate Airflow Events

The mcp-airflow-api repository provides three asynchronous MCP tools—list_event_logs, get_event_log, and all_dag_event_summary—that enable programmatic querying, detailed retrieval, and statistical aggregation of Airflow event logs through the REST API.

The mcp-airflow-api package exposes a focused suite of event log analysis capabilities designed specifically for LLM assistants and automated monitoring workflows. These asynchronous tools interface directly with Airflow's REST API endpoints to provide paginated listing, individual record inspection, and cross-DAG statistical summaries.

Available Event Log Analysis Tools

All event log analysis functionality is implemented in src/mcp_airflow_api/tools/common_tools.py and registered with the @mcp.tool() decorator, making the tools automatically discoverable by the MCP runtime.

list_event_logs

The list_event_logs function returns a paginated list of event log records with optional filtering capabilities. According to the source code in common_tools.py (lines 360-369), this tool accepts dag_id, limit, and offset parameters to construct a targeted query string for the GET /eventLogs endpoint, returning a dictionary containing the event_logs array and pagination metadata.

get_event_log

For detailed inspection of specific occurrences, get_event_log retrieves complete event payloads by their numeric identifier. The implementation (lines 371-376) issues a request to GET /eventLogs/{event_log_id} and returns the full record structure, including fields such as execution_date, task_id, event type, and any attached extra metadata.

all_dag_event_summary

The all_dag_event_summary tool (lines 378-399) provides high-level statistical analysis by aggregating up to 1,000 recent log entries. It iterates through the retrieved events to count occurrences per DAG and per event type (e.g., task_success, task_failure), returning a dictionary with total_events, unique_dags, and a detailed event_summary breakdown that helps identify systemic issues like recurring task failures.

Event Log Analysis Workflow

These tools follow a structured three-phase investigation pattern that supports both manual debugging and automated monitoring:

1. Exploration: Use list_event_logs to scan recent activity across all DAGs or constrain the view to a specific workflow using the dag_id parameter, with limit and offset controlling pagination.
2. Investigation: Drill down into suspicious entries using get_event_log to examine full execution context, including timestamps, task identifiers, and attached diagnostic data.
3. Aggregation: Apply all_dag_event_summary to identify broad patterns across your Airflow instance, such as which DAGs generate the most failures or which event types dominate your logs.

Implementation Examples

Querying Recent Events

from mcp_airflow_api.tools.common_tools import list_event_logs

# Retrieve the last 10 events for a specific DAG

logs = await list_event_logs(dag_id="example_dag", limit=10, offset=0)
print(logs["event_logs"])

Retrieving Detailed Records

from mcp_airflow_api.tools.common_tools import get_event_log

# Fetch complete payload for event ID 12345

detail = await get_event_log(event_log_id=12345)
print(f"Event type: {detail['event']}, Task: {detail.get('task_id')}")

Generating Statistical Summaries

from mcp_airflow_api.tools.common_tools import all_dag_event_summary

# Analyze patterns across all DAGs

summary = await all_dag_event_summary()
print(f"Total events analyzed: {summary['total_events']}")
print(f"Active DAGs: {summary['unique_dags']}")
for dag, counts in summary['event_summary'].items():
    print(f"{dag}: {counts}")

Integration with MCP Assistants

The interactive troubleshooting guide in src/mcp_airflow_api/mcp_main.py (lines 140-146) explicitly lists list_event_logs as the recommended command for "Deep Dive Investigation" scenarios. Because all event log analysis tools are defined as async def functions, they support concurrent execution in LLM-driven agent workflows without blocking the event loop, enabling efficient parallel monitoring across multiple DAGs.

Summary

• Three specialized tools in src/mcp_airflow_api/tools/common_tools.py provide comprehensive event log analysis capabilities: paginated listing, single-record retrieval, and statistical aggregation.
• Asynchronous architecture allows non-blocking concurrent queries against the Airflow REST API, essential for high-throughput monitoring scenarios.
• Statistical aggregation via all_dag_event_summary processes up to 1,000 recent entries to surface failure patterns and quantify DAG activity levels.
• Native MCP integration through the @mcp.tool() decorator ensures automatic discovery by LLM assistants, with usage patterns documented in mcp_main.py.

Frequently Asked Questions

How do I filter event logs for a specific DAG?

Use the dag_id parameter in list_event_logs. Pass the DAG identifier as a string to narrow results to that specific workflow, then apply limit and offset for pagination control through large result sets.

What information does the event log summary provide?

The all_dag_event_summary tool returns a dictionary containing total_events (count), unique_dags (distinct DAG identifiers), and event_summary (breakdown of event types per DAG). This structure enables quick identification of high-error workflows and activity distribution across your Airflow instance.

Are these event log analysis tools asynchronous?

Yes. All three tools (list_event_logs, get_event_log, all_dag_event_summary) are defined as async def functions in src/mcp_airflow_api/tools/common_tools.py. This design allows them to run concurrently in async MCP workflows without blocking the main execution thread.

Where are the event log tools registered in the codebase?

The tools are defined in src/mcp_airflow_api/tools/common_tools.py (lines 360-399) and exposed through the package via src/mcp_airflow_api/tools/__init__.py. Usage guidance and integration patterns are documented in the troubleshooting section of src/mcp_airflow_api/mcp_main.py.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}