MCP integration

Call tools exposed by external MCP servers from inside a Flux workflow using the mcp() client.

This page covers Flux as an MCP consumer: a workflow that connects to an external MCP server and calls the tools it exposes. If you want the other direction — Flux exposing its own workflows as MCP tools — see Running workflows → From MCP.

The entry point is mcp() from flux.tasks.mcp:

from flux.tasks.mcp import mcp

Basic usage

mcp() returns an async context manager that wraps a single MCP server. Inside the block, call discover() to retrieve the available tools as a ToolSet, then call tools by attribute name:

from flux import workflow, ExecutionContext
from flux.tasks.mcp import mcp

@workflow
async def my_workflow(ctx: ExecutionContext):
    async with mcp("http://localhost:8080/mcp", name="server") as client:
        tools = await client.discover()
        result = await tools.list_workflows()
        return result

mcp() itself does not open a connection. The context manager’s __aenter__ stores configuration only. discover() is where the actual connection happens.

Tool discovery

discover() is a Flux @task. When it runs, it connects to the MCP server, calls list_tools, and returns a ToolSet — a dict-like object where each tool is a callable Flux task:

async with mcp("http://localhost:8080/mcp", name="flux") as client:
    tools = await client.discover()

    # Attribute access — calls the tool named "get_weather"
    result = await tools.get_weather(city="London")

    # Iteration — useful for passing all tools to an agent
    for tool in tools:
        print(tool.name)

    print(f"Found {len(tools)} tools")

Tool schemas are serialized into the Flux event log at discovery time. On workflow resume, discover() replays from the log without reconnecting to the server. This is consistent with Flux’s event-sourcing model: no external calls happen during replay.

If the server’s tool list may have changed since the workflow paused — for example after a long human-in-the-loop wait — call rediscover() instead:

tools = await client.rediscover()

Each rediscover() call gets a deterministic name (mcp_{server}_rediscover_1, mcp_{server}_rediscover_2, …) so the event log stays replay-safe across multiple calls.

Pass tools to an agent

ToolSet is iterable, so you can pass all discovered tools directly to agent():

from flux.tasks.ai import agent
from flux.tasks.mcp import mcp

@workflow
async def assistant_workflow(ctx: ExecutionContext):
    async with mcp("http://localhost:8080/mcp", name="flux") as client:
        tools = await client.discover()

        assistant = await agent(
            "You are a workflow manager. Use the available tools to help the user.",
            model="ollama/llama3.2",
            tools=list(tools),
        )

        return await assistant("What workflows are available?")

To pass a subset, list the specific tool attributes:

assistant = await agent(
    "...",
    model="ollama/llama3.2",
    tools=[tools.list_workflows, tools.get_workflow_details],
)

The agent reads each tool’s name and description from its schema to build the LLM tool definitions automatically.

Authentication

Bearer token

Pass a static token, a Flux secret store key, or a callable:

from flux.tasks.mcp import mcp, bearer

# Static token
async with mcp("https://api.example.com/mcp", auth=bearer("my-token")) as client: ...

# Resolved from the Flux secret store at connection time
async with mcp("https://api.example.com/mcp", auth=bearer(secret="MCP_API_KEY")) as client: ...

# Callable — sync or async; called at each connection
async with mcp("https://api.example.com/mcp", auth=bearer(provider=get_fresh_token)) as client: ...

Tokens are resolved at connection time, not when mcp() is called. After a workflow pause of several hours, the reconnect fetches a fresh token rather than reusing a stale one.

OAuth 2.1

from flux.tasks.mcp import mcp, oauth

async with mcp(
    "https://api.example.com/mcp",
    auth=oauth(scopes=["read", "write"], client_name="My App"),
) as client:
    tools = await client.discover()

OAuth handling is delegated to FastMCP, which manages server discovery, PKCE, token exchange, and automatic refresh.

Connection modes

By default (connection="session"), one connection is shared across all tool calls within the async with block:

async with mcp("http://localhost:8080/mcp", connection="session") as client:
    tools = await client.discover()
    await tools.tool_a()   # reuses the connection
    await tools.tool_b()   # reuses the connection

For long-lived workflows where MCP calls are infrequent, use connection="per-call" to avoid holding open a connection across hours of idle time:

async with mcp("http://localhost:8080/mcp", connection="per-call") as client:
    tools = await client.discover()
    await tools.tool_a()   # opens, calls, closes
    await tools.tool_b()   # opens, calls, closes

Retries, timeouts, and caching

Pass task options at the client level; all discovered tools inherit them:

async with mcp(
    "http://localhost:8080/mcp",
    name="server",
    retry_max_attempts=3,
    retry_delay=1,
    timeout=30,
) as client:
    tools = await client.discover()
    result = await tools.some_tool(arg="value")

Override options on individual tools with with_options():

tools = await client.discover()

long_running = tools.execute_workflow_sync.with_options(timeout=120)
result = await long_running(workflow_name="heavy_job", input_data="{}")

Error handling

MCP tool failures raise ToolExecutionError, which extends Flux’s ExecutionError. Retry and fallback apply to MCP tools the same way they apply to any other task:

async with mcp("http://localhost:8080/mcp", retry_max_attempts=3, timeout=30) as client:
    tools = await client.discover()
    result = await tools.some_tool(arg="value")
    # On connection drop or server error, Flux retries up to 3 times.
    # Each retry discards the stale connection and opens a fresh one,
    # re-resolving auth in the process.

Connection errors (timeouts, refused connections) discard the current connection. The next retry opens a new one.

Connecting to multiple servers

Use one mcp() per server. Nest the context managers and handle orchestration in the workflow:

async with mcp("http://server-a:8080/mcp", name="a") as a:
    async with mcp("http://server-b:8081/mcp", name="b") as b:
        a_tools = await a.discover()
        b_tools = await b.discover()

        result_a = await a_tools.some_tool()
        result_b = await b_tools.other_tool()

Tool names in the event log are prefixed with the server name (mcp_a_some_tool, mcp_b_other_tool) so there are no collisions across servers.

Pause and resume

MCP tools work with Flux’s pause/resume without special handling:

from flux.tasks import pause

@workflow
async def approval_workflow(ctx: ExecutionContext):
    async with mcp("http://localhost:8080/mcp", name="flux") as client:
        tools = await client.discover()

        available = await tools.list_workflows()
        user_input = await pause("choose_workflow", output=available)

        # On resume, discover() and list_workflows() replay from events
        # without reconnecting. Only this call actually hits the server.
        result = await tools.execute_workflow_sync(
            workflow_name=user_input["workflow_name"],
            input_data="{}",
        )
        return result

After resume, completed tasks replay from the event log. The first new tool call triggers a lazy reconnect.

Testing

FastMCP’s in-memory transport lets you test MCP-integrated workflows without a running server:

from fastmcp import FastMCP
from flux.tasks.mcp import mcp

server = FastMCP("test")

@server.tool()
def get_weather(city: str) -> str:
    return f"Sunny in {city}"

async with mcp(server, name="test") as client:
    tools = await client.discover()
    result = await tools.get_weather(city="London")
    assert "London" in result

Pass a FastMCP instance instead of a URL string. The client uses in-process transport — no HTTP, no ports.

`mcp()` reference

def mcp(
    server: str | FastMCP,     # URL or FastMCP instance (for testing)
    *,
    auth=None,                 # bearer(...) or oauth(...)
    name: str | None = None,   # server name; defaults to hostname from URL
    connection: str = "session",  # "session" or "per-call"
    connect_timeout: int = 10,    # MCP handshake timeout in seconds
    retry_max_attempts: int = 0,  # default retries for all tools
    retry_delay: int = 1,         # initial retry delay in seconds
    retry_backoff: int = 2,       # retry backoff multiplier
    timeout: int = 0,             # default task timeout for all tools
    cache: bool = False,          # enable result caching for all tools
) -> MCPClient:

When name is not provided, the client derives it from the hostname in the URL. The derived name is used as the prefix for all event log entries (mcp_{name}_discover, mcp_{name}_{tool_name}).