MCP Server Connection¶

The MCP connection type configures access to MCP (Model Context Protocol) servers. Three transport types are supported: Streamable HTTP, SSE, and stdio.

Default Connection IDs¶

The MCPHook uses mcp_default by default.

Configuring the Connection¶

Transport (Extra field)

The transport type: http (default), sse, or stdio.

http: Streamable HTTP — the recommended transport for remote servers.
sse: Server-Sent Events — deprecated in favor of Streamable HTTP.
stdio: Run the MCP server as a subprocess communicating over stdin/stdout.

Host

The server URL. Required for http and sse transports.

Examples: http://localhost:3001/mcp, https://mcp.example.com/v1

Auth Token (Password field)

Optional authentication token for the MCP server. Sent as a static Authorization: Bearer <token> header on HTTP/SSE transports. For short-lived or minted tokens, use a token_provider instead (see below).

Command (Extra field)

The command to run for stdio transport. Required when transport is stdio.

Examples: uvx, python, node

Arguments (Extra field)

JSON array of arguments for the stdio command.

Examples: ["mcp-run-python"], ["-m", "my_mcp_server"]

Examples¶

HTTP transport (remote MCP server)

{
    "conn_type": "mcp",
    "host": "http://localhost:3001/mcp"
}

SSE transport

{
    "conn_type": "mcp",
    "host": "http://localhost:3001/sse",
    "extra": "{\"transport\": \"sse\"}"
}

Stdio transport (subprocess)

{
    "conn_type": "mcp",
    "extra": "{\"transport\": \"stdio\", \"command\": \"uvx\", \"args\": [\"mcp-run-python\"]}"
}

Stdio with custom timeout

{
    "conn_type": "mcp",
    "extra": "{\"transport\": \"stdio\", \"command\": \"python\", \"args\": [\"-m\", \"my_server\"], \"timeout\": 30}"
}

Short-lived or minted tokens¶

Some MCP endpoints require a freshly minted, short-lived token rather than a static one. For example, Snowflake managed MCP servers are best authenticated with a key-pair JWT: the private key never leaves your environment and the signed JWT expires after about an hour, so it cannot be stored as a static connection password. The same applies to OAuth / refresh tokens, Workload Identity Federation, and GitHub App installation tokens.

For these, pass a token_provider callable to MCPHook or MCPToolset instead of a static token. It is called each time the connection is established and its return value is used as the bearer token, so a fresh token is minted (and registered with secret masking so it does not leak into task logs):

from airflow.providers.common.ai.toolsets.mcp import MCPToolset


def mint_snowflake_jwt() -> str:
    # Sign a short-lived JWT from the Snowflake connection's key-pair.
    ...


toolset = MCPToolset(
    mcp_conn_id="snowflake_managed_mcp",
    token_provider=mint_snowflake_jwt,
)

token_provider is resolved in DAG code (it is a Python callable, not a stored connection field), so the signing key stays in your environment and is never baked into the serialized DAG.