MCP Server Airflow Token

Instructions

Implementation Reference

• src/airflow/dataset.py:66-70 (handler)

  Handler function that executes the 'get_dataset' tool: fetches dataset by URI from Airflow API and returns as text content.

  async def get_dataset(
      uri: str,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      response = dataset_api.get_dataset(uri=uri)
      return [types.TextContent(type="text", text=str(response.to_dict()))]

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

  Registration tuple for the 'get_dataset' tool within the list returned by get_all_functions().

  (get_dataset, "get_dataset", "Get a dataset by URI", True),

• src/main.py:10-10 (registration)

  Import of the function that provides the list of dataset tools, including get_dataset.

  from src.airflow.dataset import get_all_functions as get_dataset_functions

• src/main.py:28-28 (registration)

  Mapping APIType.DATASET to get_dataset_functions for loading tools.

  APIType.DATASET: get_dataset_functions,

• src/main.py:90-92 (registration)

  Code that iterates over functions (including get_dataset) and registers them as MCP tools using app.add_tool.

  for func, name, description, *_ in functions:
      app.add_tool(func, name=name, description=description)