get_event_log
Retrieve specific Airflow event log entries by ID to monitor workflow execution details and troubleshoot pipeline issues.
Instructions
Get a specific log entry by ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| event_log_id | Yes |
Implementation Reference
- src/airflow/eventlog.py:71-75 (handler)The main handler function for the 'get_event_log' tool. It takes an event_log_id, calls the Airflow EventLogApi, and returns the response as text content.
async def get_event_log( event_log_id: int, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: response = event_log_api.get_event_log(event_log_id=event_log_id) return [types.TextContent(type="text", text=str(response.to_dict()))] - src/airflow/eventlog.py:12-17 (registration)Module-level registration function that includes the 'get_event_log' tool in the list for MCP server registration.
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:79-92 (registration)Top-level registration loop in main.py that imports and adds 'get_event_log' via get_eventlog_functions() and app.add_tool.
logging.debug(f"Adding API: {api}") get_function = APITYPE_TO_FUNCTIONS[APIType(api)] try: functions = get_function() except NotImplementedError: continue # Filter functions for read-only mode if requested if read_only: functions = filter_functions_for_read_only(functions) for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)