get_event_logs
Retrieve and filter Airflow event log entries to monitor DAG runs, task executions, and workflow events for debugging and analysis.
Instructions
List log entries from event log
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| offset | No | ||
| order_by | No | ||
| dag_id | No | ||
| task_id | No | ||
| run_id | No | ||
| map_index | No | ||
| try_number | No | ||
| event | No | ||
| owner | No | ||
| before | No | ||
| after | No | ||
| included_events | No | ||
| excluded_events | No |
Implementation Reference
- src/airflow/eventlog.py:20-69 (handler)The main handler function for the 'get_event_logs' tool. It accepts optional filter parameters, builds a kwargs dict, calls the Airflow EventLogApi.get_event_logs, and returns the response as a list containing a TextContent object.
async def get_event_logs( limit: Optional[int] = None, offset: Optional[int] = None, order_by: Optional[str] = None, dag_id: Optional[str] = None, task_id: Optional[str] = None, run_id: Optional[str] = None, map_index: Optional[int] = None, try_number: Optional[int] = None, event: Optional[str] = None, owner: Optional[str] = None, before: Optional[datetime] = None, after: Optional[datetime] = None, included_events: Optional[str] = None, excluded_events: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: # Build parameters dictionary kwargs: Dict[str, Any] = {} if limit is not None: kwargs["limit"] = limit if offset is not None: kwargs["offset"] = offset if order_by is not None: kwargs["order_by"] = order_by if dag_id is not None: kwargs["dag_id"] = dag_id if task_id is not None: kwargs["task_id"] = task_id if run_id is not None: kwargs["run_id"] = run_id if map_index is not None: kwargs["map_index"] = map_index if try_number is not None: kwargs["try_number"] = try_number if event is not None: kwargs["event"] = event if owner is not None: kwargs["owner"] = owner if before is not None: kwargs["before"] = before if after is not None: kwargs["after"] = after if included_events is not None: kwargs["included_events"] = included_events if excluded_events is not None: kwargs["excluded_events"] = excluded_events response = event_log_api.get_event_logs(**kwargs) return [types.TextContent(type="text", text=str(response.to_dict()))] - src/airflow/eventlog.py:12-17 (registration)Local registration of the 'get_event_logs' tool as part of the list returned by get_all_functions, which provides the function, name, description, and read-only flag.
def get_all_functions() -> list[tuple[Callable, str, str, bool]]: """Return list of (function, name, description, is_read_only) tuples for registration.""" return [ (get_event_logs, "get_event_logs", "List log entries from event log", True), (get_event_log, "get_event_log", "Get a specific log entry by ID", True), ] - src/main.py:90-92 (registration)Central MCP tool registration loop in main.py where functions from eventlog.get_all_functions (including get_event_logs) are registered by calling app.add_tool with the function, name, and description.
for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)