MCP Server for Apache Airflow

Instructions

Implementation Reference

• src/airflow/dataset.py:73-103 (handler)

  The main handler function for the 'get_dataset_events' tool. It accepts optional parameters, builds a kwargs dict, calls dataset_api.get_dataset_events, and returns the response as TextContent.

  async def get_dataset_events(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      order_by: Optional[str] = None,
      dataset_id: Optional[int] = None,
      source_dag_id: Optional[str] = None,
      source_task_id: Optional[str] = None,
      source_run_id: Optional[str] = None,
      source_map_index: Optional[int] = 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 dataset_id is not None:
          kwargs["dataset_id"] = dataset_id
      if source_dag_id is not None:
          kwargs["source_dag_id"] = source_dag_id
      if source_task_id is not None:
          kwargs["source_task_id"] = source_task_id
      if source_run_id is not None:
          kwargs["source_run_id"] = source_run_id
      if source_map_index is not None:
          kwargs["source_map_index"] = source_map_index
  
      response = dataset_api.get_dataset_events(**kwargs)
      return [types.TextContent(type="text", text=str(response.to_dict()))]

• src/airflow/dataset.py:11-39 (registration)

  The get_all_functions() returns a list of tuples for tool registration, including the one for 'get_dataset_events' at line 16.

  def get_all_functions() -> list[tuple[Callable, str, str, bool]]:
      """Return list of (function, name, description, is_read_only) tuples for registration."""
      return [
          (get_datasets, "get_datasets", "List datasets", True),
          (get_dataset, "get_dataset", "Get a dataset by URI", True),
          (get_dataset_events, "get_dataset_events", "Get dataset events", True),
          (create_dataset_event, "create_dataset_event", "Create dataset event", False),
          (get_dag_dataset_queued_event, "get_dag_dataset_queued_event", "Get a queued Dataset event for a DAG", True),
          (get_dag_dataset_queued_events, "get_dag_dataset_queued_events", "Get queued Dataset events for a DAG", True),
          (
              delete_dag_dataset_queued_event,
              "delete_dag_dataset_queued_event",
              "Delete a queued Dataset event for a DAG",
              False,
          ),
          (
              delete_dag_dataset_queued_events,
              "delete_dag_dataset_queued_events",
              "Delete queued Dataset events for a DAG",
              False,
          ),
          (get_dataset_queued_events, "get_dataset_queued_events", "Get queued Dataset events for a Dataset", True),
          (
              delete_dataset_queued_events,
              "delete_dataset_queued_events",
              "Delete queued Dataset events for a Dataset",
              False,
          ),
      ]