get_event_log
Retrieve a specific Apache Airflow log entry by its ID to monitor workflow execution details and troubleshoot pipeline issues.
Instructions
Get a specific log entry by ID
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| event_log_id | Yes |
Input Schema (JSON Schema)
{
"properties": {
"event_log_id": {
"title": "Event Log Id",
"type": "integer"
}
},
"required": [
"event_log_id"
],
"type": "object"
}
Implementation Reference
- src/airflow/eventlog.py:71-75 (handler)The main handler function implementing the 'get_event_log' tool. It takes an event_log_id, calls the Airflow EventLogApi to retrieve the log entry, converts the response to a dictionary string, and returns it wrapped in MCP TextContent format.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)Local registration of the 'get_event_log' tool within the module's get_all_functions(), which returns tuples of (function, name, description, read_only) used by the main.py to register tools with the MCP server.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-91 (registration)Global tool registration loop in main.py where tools from eventlog.get_all_functions (via APITYPE_TO_FUNCTIONS[APIType.EVENTLOG]) are added to the MCP app using app.add_tool.for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)