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

> Analyze mcp-airflow-api event logs programmatically. Query, retrieve, and aggregate Airflow events with list event logs, get event log, and all dag event summary tools.

- Repository: [JungJungIn/mcp-airflow-api](https://github.com/call518/mcp-airflow-api)
- Tags: tutorial
- Published: 2026-02-26

---

**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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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

```python
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

```python
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

```python
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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/src/mcp_airflow_api/tools/common_tools.py) (lines 360-399) and exposed through the package via [`src/mcp_airflow_api/tools/__init__.py`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/src/mcp_airflow_api/mcp_main.py).