get_event_log
Retrieve specific Apache Airflow event log entries by ID to monitor system activities and troubleshoot workflow execution 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 to retrieve the log entry, and returns it as MCP TextContent.
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:14-17 (registration)The registration tuple for the 'get_event_log' tool within the module's get_all_functions(), which lists available tools for this module.
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:12-12 (registration)Import of the eventlog module's get_all_functions in the main.py, which collects tools from all modules.
from src.airflow.eventlog import get_all_functions as get_eventlog_functions - src/main.py:95-96 (registration)The loop in main() that registers all collected tools (including get_event_log) with the MCP app using Tool.from_function.
for func, name, description, *_ in functions: app.add_tool(Tool.from_function(func, name=name, description=description)) - src/airflow/eventlog.py:9-9 (helper)Initialization of the EventLogApi client used by the get_event_log handler.
event_log_api = EventLogApi(api_client)