How the MCP Airflow API Server Handles Connection Management to Airflow

The MCP server maintains a single, long-lived aiohttp.ClientSession with configurable connection pooling that is shared across all Airflow API requests, ensuring efficient connection reuse and centralized authentication handling.

The call518/mcp-airflow-api repository implements a robust connection management strategy for the Model Context Protocol (MCP) server that communicates with Apache Airflow's REST API. By leveraging a global session pattern with aiohttp, the server minimizes connection overhead while supporting both Airflow API v1 and v2 endpoints through a unified interface.

Global Session Architecture with Connection Pooling

The server utilizes a single, global aiohttp.ClientSession instantiated in src/mcp_airflow_api/functions.py rather than creating new connections per request. This architectural decision eliminates TCP handshake overhead and enables HTTP Keep-Alive for persistent connections to the Airflow backend.

TCPConnector Configuration

The session initialization configures a TCPConnector with specific limits to balance concurrency and resource utilization:

# From src/mcp_airflow_api/functions.py

connector = aiohttp.TCPConnector(
    limit=10,              # Total connections across all hosts

    limit_per_host=5,      # Maximum connections to single Airflow instance

    keepalive_timeout=30   # Seconds to keep idle connections alive

)

These settings ensure that the MCP server maintains up to 10 total concurrent connections with a maximum of 5 dedicated to the Airflow host, while keeping idle sockets available for 30 seconds to facilitate rapid subsequent requests.

Session Lifecycle Management

The get_airflow_session() function in src/mcp_airflow_api/functions.py implements lazy initialization, creating the session only upon first use and returning the cached instance for all subsequent calls. When the server terminates, close_airflow_session() explicitly closes the global session to release file descriptors and network resources properly.

Centralized Authentication and Request Routing

All Airflow API interactions flow through the airflow_request() wrapper function, which handles connection management to Airflow transparently for both API versions.

Version-Aware URL Construction

The construct_api_url() function dynamically builds endpoints by combining the base URL with the appropriate API version prefix (v1 or v2):

# Conceptual implementation from src/mcp_airflow_api/functions.py

def construct_api_url(path: str, version: str = None) -> str:
    base = BASE_URL.rstrip('/')
    ver = version or AIRFLOW_API_VERSION
    return f"{base}/{ver}{path}"

This approach allows the same underlying ClientSession to communicate with both legacy Airflow 2.x (v1) and modern Airflow 3.x (v2) endpoints without requiring separate connection pools.

Authentication Header Injection

Before delegating to the shared session, airflow_request() injects the appropriate authentication credentials. For Airflow API v1, it applies Basic Auth headers, while v2 endpoints receive JWT Bearer tokens. This centralized authentication ensures that security credentials are consistently applied across all tool operations without duplicate code in individual endpoint handlers.

Implementation in the MCP Server

During server startup in src/mcp_airflow_api/mcp_main.py, the MCP instance initializes once and loads tools from v1_tools.py, v2_tools.py, and common_tools.py. Each tool implementation ultimately calls airflow_request(), meaning every DAG listing, task instance query, or variable update reuses the same pooled connection.

Practical Code Examples

Reusing the Shared Session for DAG Queries

from mcp_airflow_api.functions import airflow_request

async def list_active_dags():
    """Fetch DAGs using the shared connection pool."""
    response = await airflow_request("GET", "/dags?limit=10&only_active=true")
    response.raise_for_status()
    return response.json()

This call automatically utilizes the global session maintained by get_airflow_session(), benefiting from connection reuse and Keep-Alive optimization.

Explicit Version Selection

from mcp_airflow_api.functions import airflow_request_v2

async def fetch_assets():
    """Force v2 API usage for Airflow 3.x features."""
    response = await airflow_request_v2("GET", "/assets")
    response.raise_for_status()
    return response.json()

While airflow_request_v2() temporarily switches the API version context, it continues using the same underlying ClientSession instance for efficient connection management to Airflow.

Graceful Shutdown Handling

from mcp_airflow_api.functions import close_airflow_session

async def shutdown_server():
    """Clean up connection resources on server termination."""
    await close_airflow_session()

Explicitly calling close_airflow_session() ensures that the aiohttp connector releases all sockets, preventing resource leaks during testing or server restarts.

Summary

• Single Global Session: The server maintains one aiohttp.ClientSession in src/mcp_airflow_api/functions.py, eliminating per-request connection overhead.
• Configurable Pooling: The TCPConnector limits connections to 10 total and 5 per host with 30-second Keep-Alive timeouts.
• Centralized Routing: All tools use airflow_request() which handles URL construction, authentication, and session management uniformly.
• Version Agnostic: The same connection pool serves both Airflow API v1 and v2 endpoints through dynamic URL construction.
• Clean Resource Management: close_airflow_session() provides explicit lifecycle control for proper resource cleanup.

Frequently Asked Questions

How does the MCP server prevent connection exhaustion when handling concurrent requests?

The TCPConnector configured in src/mcp_airflow_api/functions.py enforces hard limits of 10 total connections and 5 connections per Airflow host. When these limits are reached, additional requests wait for available connections in the pool rather than opening new sockets, preventing file descriptor exhaustion and maintaining stable connection management to Airflow under load.

Can the server communicate with multiple Airflow instances simultaneously?

Yes. While the server maintains a single ClientSession, the limit_per_host=5 parameter specifically restricts connections per individual Airflow host. If configured with different base URLs, the same session can maintain separate connection pools to multiple Airflow instances, adhering to the per-host limits for each backend.

What happens to active connections when the MCP server shuts down?

The close_airflow_session() function explicitly closes the global aiohttp.ClientSession, which terminates all active connections in the pool and releases underlying TCP sockets. This graceful shutdown prevents connection leaks and ensures that the Airflow server receives proper TCP FIN packets rather than experiencing abrupt connection drops.

Does using a shared session impact authentication security between API versions?

No. Although the session is shared, the airflow_request() wrapper in src/mcp_airflow_api/functions.py injects authentication headers immediately before each request dispatch. This means v1 requests receive Basic Auth headers while v2 requests receive Bearer tokens, ensuring that credentials remain isolated and secure despite using the same underlying TCP connections.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}