[Feature] Dynamic Runtime Loading and Refresh of MCP Tools and Prompts

[yanand0909]

GitHub Issue: #458

Motivation

Current MCP Resource Discovery

The Flink Agents framework currently discovers MCP tools and prompts at compile time during AgentPlan construction. In AgentPlan.extractJavaMCPServer(), the framework:

1. Instantiates the MCP server via JavaResourceProvider.provide()
2. Calls listTools() and listPrompts() over the network
3. Serializes discovered resources as JavaSerializableResourceProvider instances
4. Closes the MCP server connection

// Current behavior in AgentPlan.extractJavaMCPServer() (plan/src/.../AgentPlan.java:407)
JavaResourceProvider provider = new JavaResourceProvider(name, MCP_SERVER, descriptor);
addResourceProvider(provider);
Object mcpServer = provider.provide(null);  // Instantiates MCP server NOW

// Discovers tools over the network at compile time
Iterable<? extends SerializableResource> tools =
    (Iterable<? extends SerializableResource>) listToolsMethod.invoke(mcpServer);

// ... registers each tool as a static resource provider ...

closeMethod.invoke(mcpServer);  // Closes connection immediately

The Problem

This compile-time discovery approach has several limitations:

1. Build-time server dependency: The MCP server must be reachable during AgentPlan construction. If the server is down, the build fails — even though the agent won't run until later.

2. Static capabilities: Tools and prompts are frozen at compile time. If an MCP server adds, removes, or updates its tools, the agent cannot see the changes without recompilation.

3. No connection reuse: The MCP server connection is opened for discovery and immediately closed. At runtime, a new connection must be established for tool calls, duplicating work.

4. No graceful degradation: A transient MCP server outage during compilation is a hard failure with no recovery path.

Proposed Solution

We propose a two-part solution:

1. Runtime Discovery — Defer MCP tool/prompt discovery from compile-time to operator startup (ActionExecutionOperator.open()). This decouples the build from MCP server availability and enables connection reuse.

2. Dynamic Refresh — Allow MCP tools and prompts to be refreshed at runtime, so agents can adapt to changes in MCP server capabilities without redeployment.

---

Public Interface

MCPServer Builder Additions

New configuration options on the existing MCPServer.Builder:

MCPServer server = MCPServer.builder("http://api.example.com/mcp")
    .auth(new BearerTokenAuth(System.getenv("API_TOKEN")))
    .timeout(Duration.ofSeconds(30))
    // Startup behavior
    .failFastOnStartup(true)       // default: true — fail operator if server unreachable
    // Dynamic refresh configuration
    .autoRefresh(true)              // default: false — enable periodic refresh
    .refreshInterval(Duration.ofMinutes(5))  // default: 5 minutes
    .build();

New Public API on MCPServer

public class MCPServer extends Resource {
    RefreshResult refreshNow();

    // Automatic refresh lifecycle
    void startAutoRefresh();
    void stopAutoRefresh();
    boolean isAutoRefreshRunning();

    // Observability
    long getLastRefreshTime();
}

### Key Points

- The `@MCPServer` annotation API is **unchanged**. Existing user code compiles and runs without modification.
- Discovery moves transparently from plan compilation to operator startup.
- Dynamic refresh is **opt-in** via `autoRefresh(true)`. Without it, tools are discovered once at startup (current behavior, just deferred).
- `failFastOnStartup(true)` (default) preserves existing fail-fast semantics. Setting it to `false` enables graceful degradation — the operator starts without MCP tools if the server is unreachable.

---

## Implementation Details

### Part 1: Runtime Discovery

#### 1.1 New MCPServerResourceProvider

A new `ResourceProvider` that stores only MCP server configuration at compile time and defers instantiation and discovery to runtime.

**File:** `plan/src/main/java/.../resourceprovider/MCPServerResourceProvider.java` (NEW)

```java
public class MCPServerResourceProvider extends ResourceProvider {
    private final ResourceDescriptor descriptor;
    private transient MCPServer serverInstance;
    private transient Map<String, MCPTool> discoveredTools;
    private transient Map<String, MCPPrompt> discoveredPrompts;

    public MCPServerResourceProvider(String name, ResourceDescriptor descriptor) {
        super(name, ResourceType.MCP_SERVER);
        this.descriptor = descriptor;
    }

    @Override
    public Resource provide(BiFunction<String, ResourceType, Resource> getResource) {
        if (serverInstance == null) {
            serverInstance = createMCPServer();
        }
        return serverInstance;
    }

    public void discoverResources() throws Exception {
        if (serverInstance == null) { provide(null); }

        discoveredTools = new HashMap<>();
        for (MCPTool tool : serverInstance.listTools()) {
            discoveredTools.put(tool.getName(), tool);
        }

        discoveredPrompts = new HashMap<>();
        for (MCPPrompt prompt : serverInstance.listPrompts()) {
            discoveredPrompts.put(prompt.getName(), prompt);
        }
    }

    public void close() throws Exception {
        if (serverInstance != null) {
            serverInstance.close();
            serverInstance = null;
        }
    }
}

1.2 Modify AgentPlan.extractJavaMCPServer()

File: plan/src/main/java/.../AgentPlan.java (lines 407-452)

Remove all MCP server instantiation, listTools()/listPrompts() calls, and close(). Replace with a simple provider registration:

private void extractJavaMCPServer(Method method) throws Exception {
    String name = method.getName();
    ResourceDescriptor descriptor = (ResourceDescriptor) method.invoke(null);
    descriptor = new ResourceDescriptor(
        descriptor.getModule(),
        JAVA_MCP_SERVER_CLASS_NAME,
        new HashMap<>(descriptor.getInitialArguments()));

    // Store configuration only — no network calls
    MCPServerResourceProvider provider = new MCPServerResourceProvider(name, descriptor);
    addResourceProvider(provider);
}

Impact: AgentPlan construction drops from ~1-2s (MCP round-trip) to <10ms. Builds no longer depend on MCP server availability.

1.3 Runtime Discovery in ActionExecutionOperator

File: runtime/src/main/java/.../operator/ActionExecutionOperator.java

Add MCP resource discovery during open() and cleanup during close():

@Override
public void open() throws Exception {
    super.open();
    // ... existing initialization ...
    initMCPResources();
}

private void initMCPResources() throws Exception {
    Map<String, ResourceProvider> mcpProviders =
        agentPlan.getResourceProviders().get(ResourceType.MCP_SERVER);
    if (mcpProviders == null) return;

    for (var entry : mcpProviders.entrySet()) {
        if (!(entry.getValue() instanceof MCPServerResourceProvider)) continue;
        MCPServerResourceProvider provider = (MCPServerResourceProvider) entry.getValue();

        try {
            provider.discoverResources();

            for (var tool : provider.getDiscoveredTools().entrySet()) {
                agentPlan.cacheResource(tool.getKey(), ResourceType.TOOL, tool.getValue());
            }
            for (var prompt : provider.getDiscoveredPrompts().entrySet()) {
                agentPlan.cacheResource(prompt.getKey(), ResourceType.PROMPT, prompt.getValue());
            }
        } catch (Exception e) {
            if (provider.isFailFastOnStartup()) {
                throw new RuntimeException(
                    "MCP discovery failed for: " + entry.getKey(), e);
            }
            LOG.warn("Continuing without MCP server '{}' — tools unavailable",
                entry.getKey());
        }
    }
}

1.4 AgentPlan Cache Helper

File: plan/src/main/java/.../AgentPlan.java

/** Cache a resource directly (used by runtime MCP discovery). */
public void cacheResource(String name, ResourceType type, Resource resource) {
    resourceCache
        .computeIfAbsent(type, k -> new ConcurrentHashMap<>())
        .put(name, resource);
}

1.5 Serialization Support

Files: ResourceProviderJsonSerializer.java, ResourceProviderJsonDeserializer.java

Add serialization/deserialization cases for MCPServerResourceProvider that persist only the ResourceDescriptor (no runtime state):

// Serializer
if (value instanceof MCPServerResourceProvider) {
    gen.writeStringField("type", "mcp_server");
    gen.writeFieldName("descriptor");
    serializerProvider.findValueSerializer(ResourceDescriptor.class)
        .serialize(provider.getDescriptor(), gen, serializerProvider);
}

// Deserializer
case "mcp_server":
    ResourceDescriptor desc = parser.readValueAs(ResourceDescriptor.class);
    return new MCPServerResourceProvider(name, desc);

---

Part 2: Dynamic Refresh

Once runtime discovery is in place, dynamic refresh builds on the same infrastructure. We propose a phased approach with multiple strategies.

2.1 Periodic Polling (Phase 1)

A background daemon thread periodically calls refreshNow():

public void startAutoRefresh() {
    if (!autoRefreshEnabled || refreshScheduler != null) return;

    refreshScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "MCP-Refresh-" + endpoint);
        t.setDaemon(true);
        return t;
    });

    refreshScheduler.scheduleAtFixedRate(
        () -> {
            try { refreshNow(); }
            catch (Exception e) { LOG.warn("Periodic refresh failed for {}", endpoint, e); }
        },
        refreshInterval.toMillis(),
        refreshInterval.toMillis(),
        TimeUnit.MILLISECONDS);
}

Flink Integration: ActionExecutionOperator.open() calls startAutoRefresh() on each MCP server after initial discovery. close() calls stopAutoRefresh().

2.3 Thread Safety

Tool reads are frequent (every tool call); refreshes are infrequent. We use StampedLock for optimal read-heavy concurrency:

private final StampedLock lock = new StampedLock();

public MCPTool getTool(String name) {
    // Optimistic read — no lock contention on the happy path
    long stamp = lock.tryOptimisticRead();
    MCPTool tool = toolsCache.get(name);

    if (!lock.validate(stamp)) {
        // Fallback to pessimistic read
        stamp = lock.readLock();
        try {
            tool = toolsCache.get(name);
        } finally {
            lock.unlockRead(stamp);
        }
    }
    return tool;
}

public void updateTools(List<MCPTool> newTools) {
    long stamp = lock.writeLock();
    try {
        toolsCache.clear();
        newTools.forEach(t -> toolsCache.put(t.getName(), t));
    } finally {
        lock.unlockWrite(stamp);
    }
}

2.4 Graceful Degradation

On refresh failures, the server falls back to the last known good state:

private Map<String, MCPTool> fallbackTools;  // Last successful discovery
private int consecutiveRefreshFailures = 0;

private void handleRefreshFailure(Exception e) {
    consecutiveRefreshFailures++;
    if (consecutiveRefreshFailures >= maxConsecutiveFailures) {
        connectionState = ConnectionState.FAILED;
        // Restore fallback
        toolsCache.clear();
        toolsCache.putAll(fallbackTools);
    } else {
        connectionState = ConnectionState.DEGRADED;
        // Keep using current cache
    }
}

2.5 Operator-Level vs Task-Manager-Level Refresh

Approach	Threads	Isolation	Complexity
Operator-level (recommended)	One per operator per MCP server	Full isolation between operators	Simple
Task-manager-level	One shared thread	Shared state across operators	Requires global registry

We recommend operator-level refresh for simplicity and isolation. Each ActionExecutionOperator manages the refresh lifecycle of its own MCP servers.

---

Future Work

Event-Driven Refresh

If MCP server notifications become part of the MCP specification, subscribe to server-sent events for near-instant tool updates:

MCPServer server = MCPServer.builder("http://api.example.com/mcp")
    .eventDrivenRefresh(true)  // Subscribe to SSE notifications
    .build();

server.addChangeListener(type -> {
    LOG.info("MCP resources updated: {}", type);
});

This would eliminate polling overhead entirely but requires MCP server-side support. The implementation should fall back to polling if the SSE connection drops.

TTL-Based Lazy Refresh

An alternative cache strategy where tools are refreshed lazily on access after a configurable TTL expires. Good for development environments with sporadic tool usage — avoids background threads while still ensuring freshness.

Advanced Resilience

• Circuit breaker pattern for failing MCP servers (prevent retry storms)
• Tool versioning and schema validation (detect incompatible tool signature changes)
• Metrics export via Flink's metrics system for Prometheus/Grafana dashboards

Python MCP Server Support

Extend runtime discovery to Python MCP servers using the existing Python client.

Other Agents approach

• Lanchain uses polling approach for dynamically loading tools and prompts.
• Claude desktop or Claude code uses Server sent notifications to refresh the list of available tools

[xintongsong]

Thanks for preparing this, @yanand0909.

There seems to be an overlap of efforts with @weiqingy #549. You guys may exchange ideas and figure out how to collaborate on this.

Concerning the design doc, I had a brief look at the Proposed Solution and Public Interface parts, but not the Implementation Details. A few comments so far:

1. Why do we need the option failFastOnStartup for a MCP Server? In which case would the user want to not fail the job if MCP Server is unreachable?
2. MCPServer is not a public API. Users should neither implement it, nor directly call any methods on it. They are defined via the @MCPServer annotation and resource descriptors.

[weiqingy]

Thanks for flagging the overlap, @xintongsong. I also just noticed that there's an existing issue #458 that covers the same feature request. I'll close my issue #549 in favor of this discussion and the original issue #458 to consolidate the effort.

[wenjin272]

Hi, @yanand0909 , thanks for preparing this document. The overall design looks good to me, including runtime discovering and dynamic refreshing.

I have some comments about the details.

A new ResourceProvider that stores only MCP server configuration at compile time and defers instantiation and discovery to runtime.

I think maybe we don't need to introduce a new type MCPServerResourceProvider. Because JavaResourceProvider or PythonResourceProvider stores the same content with MCPServerResourceProvider.

The difference between JavaResourceProvider and MCPServerResourceProvider is MCPServerResourceProvider provide a extra method discoverResources. I think we can convert this method to a util static method.

The benefit is we don't need to modify the SerDe for resource provider.

Add MCP resource discovery during open() and cleanup during close():

Currently the MCPTool and MCPPrompt are both SerializableResource. If we discovery and construct them during open, they don't need to be json serializable any more, which means MCPServer don't need to be json serializable any more. This allows we remove the JsonCreator of MCPServer and remove some unnecessary fields.

1.4 AgentPlan Cache Helper

In #548, we have extracted ResourceCache. So this code may need move to ResourceCache.

Part 2: Dynamic Refresh

Currently, user must register all the names of tools including mcp tools when declaring a chat model setup, which results in the inability to support scenarios where MCP tools are dynamically added or removed at runtime. After we support dynamic refresh of mcp server, we may also support register a mcp server to chat model setup, rather than only support being precise down to each tool's name.

[yanand0909]

Thanks @wenjin272 for taking a look, I think we can make JavaResourceProvider work and we won't need to add new MCPServerResourceProvider. All other points makes sense and we can make those changes as followup.