Agent Guidelines for astro-airflow-mcp

Table of Contents generated with DocToc

• Agent Guidelines for astro-airflow-mcp
  • Architecture
  • Code Conventions
    • HTTP Client
    • Adapter Pattern
    • MCP Tools
    • Error Handling
  • Airflow Version Differences
  • Testing
  • Files to Know

Agent Guidelines for astro-airflow-mcp

Guidelines for AI coding assistants contributing to this repository.

Architecture

MCP server for Apache Airflow with adapter pattern for version compatibility:

src/astro_airflow_mcp/
├── server.py          # Core infrastructure: config, auth, adapter management (FastMCP)
├── tools/             # MCP tool implementations (grouped by domain)
│   ├── dag.py         # DAG management (get, list, source, stats, pause, unpause)
│   ├── task.py        # Task management (get, list, instance, logs, clear)
│   ├── dag_run.py     # DAG run management (list, get, trigger, trigger_and_wait)
│   ├── asset.py       # Asset/dataset tools (list, events, upstream events)
│   ├── admin.py       # Admin tools (connections, variables, pools, plugins, providers, config, version)
│   └── diagnostic.py  # Diagnostic tools (warnings, errors, explore, diagnose, health)
├── resources.py       # MCP resources (read-only endpoints)
├── prompts.py         # MCP prompts (guided workflows)
├── adapters/
│   ├── base.py        # Abstract adapter interface
│   ├── airflow_v2.py  # Airflow 2.x API (/api/v1)
│   └── airflow_v3.py  # Airflow 3.x API (/api/v2)
├── models.py          # Pydantic models (documentation/type reference)
└── plugin.py          # Airflow 3.x plugin integration

Code Conventions

HTTP Client

Use httpx, not requests. All HTTP calls should use httpx.Client and pass self._verify for SSL configuration:

# Good
with httpx.Client(timeout=30.0, verify=self._verify) as client:
    response = client.get(url, headers=headers)

# Bad
response = requests.get(url, headers=headers)

Adapter Pattern

All Airflow API calls go through adapters. Never call Airflow API directly from MCP tools:

# Good - use adapter
adapter = _get_adapter()
data = adapter.list_dags(limit=100)

# Bad - direct API call
response = httpx.get(f"{url}/api/v2/dags")

When adding new API endpoints:

1. Add abstract method to adapters/base.py
2. Implement in both airflow_v2.py and airflow_v3.py
3. Handle API differences (field names, paths, availability)

MCP Tools

Tools need descriptive docstrings for AI discovery:

@mcp.tool()
def get_dag_details(dag_id: str) -> str:
    """Get detailed information about a specific DAG.

    Use this tool when the user asks about:
    - "Show me details for DAG X"
    - "What's the schedule for DAG Y?"

    Args:
        dag_id: The ID of the DAG

    Returns:
        JSON with DAG metadata
    """

Error Handling

Adapters handle missing endpoints gracefully:

try:
    return self._call("newEndpoint")
except NotFoundError:
    return self._handle_not_found(
        "newEndpoint",
        alternative="Use alternativeEndpoint instead"
    )

Airflow Version Differences

Testing

• Mock adapters, not HTTP libraries
• Unit tests: tests/
• Integration tests: tests/integration/ (require running Airflow)

# Good - mock adapter (patch in the tool module that imports it)
mock_adapter = mocker.Mock()
mock_adapter.list_dags.return_value = {"dags": [...]}
mocker.patch("astro_airflow_mcp.tools.dag._get_adapter", return_value=mock_adapter)

# Bad - mock HTTP library
mocker.patch("httpx.Client")

Files to Know

• server.py: All MCP tools defined here
• adapters/base.py: Interface contract for both versions
• models.py: Type reference (not used for validation currently)
• docker-compose.test.yml: Test against real Airflow instances