MCP Server for Apache Airflow

Instructions

Implementation Reference

• src/airflow/plugin.py:18-41 (handler)

  The main handler function for the 'get_plugins' MCP tool. It accepts optional limit and offset parameters, calls the Airflow PluginApi.get_plugins, and returns the response as a TextContent list.

  async def get_plugins(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get a list of loaded plugins.
  
      Args:
          limit: The numbers of items to return.
          offset: The number of items to skip before starting to collect the result set.
  
      Returns:
          A list of loaded plugins.
      """
      # Build parameters dictionary
      kwargs: Dict[str, Any] = {}
      if limit is not None:
          kwargs["limit"] = limit
      if offset is not None:
          kwargs["offset"] = offset
  
      response = plugin_api.get_plugins(**kwargs)
      return [types.TextContent(type="text", text=str(response.to_dict()))]

• src/airflow/plugin.py:11-16 (registration)

  The get_all_functions helper that provides the registration tuple (function, name, description, read_only) for the 'get_plugins' tool, imported and used in main.py to register the tool with the MCP server.

  def get_all_functions() -> list[tuple[Callable, str, str, bool]]:
      """Return list of (function, name, description, is_read_only) tuples for registration."""
      return [
          (get_plugins, "get_plugins", "Get a list of loaded plugins", True),
      ]