This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 0c2388acc [api][plan][java][python] Fix MCP server declarations in
YAML (#1303)
0c2388acc is described below
commit 0c2388acc84a8620bfa0fc97b24fe5ede869b26d
Author: Yi Zhang <[email protected]>
AuthorDate: Wed Oct 21 20:18:27 2026 +0800
[api][plan][java][python] Fix MCP server declarations in YAML (#1303)
Generated-by: Codex CLI 0.158.0-alpha.2.1
(mode-ed9921cee0ec46e59a12b1b9e2d913b5)
---
.../org/apache/flink/agents/api/yaml/Aliases.java | 6 ++++
.../apache/flink/agents/api/yaml/YamlLoader.java | 3 +-
.../apache/flink/agents/api/yaml/AliasesTest.java | 6 ++++
.../agents/api/yaml/YamlLoaderBuildAgentsTest.java | 35 ++++++++++++++++++++++
.../agents/api/yaml/YamlLoaderBuildersTest.java | 21 ++++++++++++-
docs/content/docs/development/create-agent/yaml.md | 10 +++++--
.../org/apache/flink/agents/plan/AgentPlan.java | 18 +++++++----
.../resourceprovider/PythonResourceProvider.java | 7 -----
.../apache/flink/agents/plan/AgentPlanTest.java | 26 ++++++++++++++++
.../PythonResourceProviderTest.java | 22 ++++++++++++++
python/flink_agents/api/yaml/aliases.py | 5 ++++
python/flink_agents/api/yaml/tests/test_aliases.py | 7 +++++
python/flink_agents/api/yaml/tests/test_loader.py | 28 +++++++++++++++++
13 files changed, 177 insertions(+), 17 deletions(-)
diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/Aliases.java
b/api/src/main/java/org/apache/flink/agents/api/yaml/Aliases.java
index da1ce5a36..b36c18311 100644
--- a/api/src/main/java/org/apache/flink/agents/api/yaml/Aliases.java
+++ b/api/src/main/java/org/apache/flink/agents/api/yaml/Aliases.java
@@ -172,6 +172,12 @@ public final class Aliases {
vsPython.put("mem0",
ResourceName.VectorStore.Python.MEM0_VECTOR_STORE);
ca.put(ResourceType.VECTOR_STORE, buildLangBuckets(vsJava, vsPython));
+ // MCP_SERVER
+ ca.put(
+ ResourceType.MCP_SERVER,
+ buildLangBuckets(
+ Map.of(), Map.of("mcp",
"flink_agents.integrations.mcp.mcp.MCPServer")));
+
CLAZZ_ALIASES = Collections.unmodifiableMap(ca);
}
diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
index 28e5d8ec8..5989d0370 100644
--- a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
+++ b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
@@ -132,7 +132,8 @@ public final class YamlLoader {
ResourceType.DECISION_MODEL_CONNECTION,
ResourceType.EMBEDDING_MODEL,
ResourceType.EMBEDDING_MODEL_CONNECTION,
- ResourceType.VECTOR_STORE)
+ ResourceType.VECTOR_STORE,
+ ResourceType.MCP_SERVER)
.contains(resourceType)) {
throw new IllegalArgumentException(
"type='python' is not supported for " +
resourceType.getValue());
diff --git
a/api/src/test/java/org/apache/flink/agents/api/yaml/AliasesTest.java
b/api/src/test/java/org/apache/flink/agents/api/yaml/AliasesTest.java
index a2b628739..b4cf83c5d 100644
--- a/api/src/test/java/org/apache/flink/agents/api/yaml/AliasesTest.java
+++ b/api/src/test/java/org/apache/flink/agents/api/yaml/AliasesTest.java
@@ -65,6 +65,12 @@ class AliasesTest {
assertThat(fqn).isEqualTo(ResourceName.ChatModel.Python.OLLAMA_CONNECTION);
}
+ @Test
+ void clazzAliasCoversPythonMCPServer() {
+ String fqn = Aliases.resolveClazz("mcp", ResourceType.MCP_SERVER,
Language.PYTHON);
+
assertThat(fqn).isEqualTo("flink_agents.integrations.mcp.mcp.MCPServer");
+ }
+
@Test
void clazzAliasCoversChatModelJavaGeminiAndAzureOpenAI() {
String geminiConn =
diff --git
a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildAgentsTest.java
b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildAgentsTest.java
index c458ccf61..2550ee5fd 100644
---
a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildAgentsTest.java
+++
b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildAgentsTest.java
@@ -158,6 +158,41 @@ class YamlLoaderBuildAgentsTest {
.containsExactly(DecisionResponseEvent.EVENT_TYPE);
}
+ @Test
+ void mcpServersSupportAgentAndSharedPythonResources(@TempDir Path tmp)
throws Exception {
+ Path file = tmp.resolve("mcp.yaml");
+ Files.writeString(
+ file,
+ "mcp_servers:\n"
+ + " - name: shared_mcp\n"
+ + " clazz: mcp\n"
+ + " endpoint: http://127.0.0.1:8000/mcp\n";
+ + "agents:\n"
+ + " - name: a\n"
+ + " mcp_servers:\n"
+ + " - name: local_mcp\n"
+ + " clazz: mcp\n"
+ + " type: python\n"
+ + " endpoint: http://127.0.0.1:8001/mcp\n";);
+
+ LoadedFile out = YamlLoader.buildAgents(file);
+ for (ResourceDescriptor descriptor :
+ new ResourceDescriptor[] {
+ (ResourceDescriptor)
+
out.getSharedResources().get(ResourceType.MCP_SERVER).get("shared_mcp"),
+ (ResourceDescriptor)
+ out.getAgents()
+ .get("a")
+ .getResources()
+ .get(ResourceType.MCP_SERVER)
+ .get("local_mcp")
+ }) {
+
assertThat(descriptor).isInstanceOf(PythonResourceDescriptor.class);
+
assertThat(descriptor.getModule()).isEqualTo("flink_agents.integrations.mcp.mcp");
+ assertThat(descriptor.getClazz()).isEqualTo("MCPServer");
+ }
+ }
+
@Test
void sharedSectionsExposedSeparatelyFromAgents() {
LoadedFile out =
YamlLoader.buildAgents(FIXTURES.resolve("with_shared.yaml"));
diff --git
a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java
b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java
index 860727bee..7063075f3 100644
---
a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java
+++
b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.agents.api.chat.messages.MessageRole;
import org.apache.flink.agents.api.function.JavaFunction;
import org.apache.flink.agents.api.function.PythonFunction;
import org.apache.flink.agents.api.prompt.Prompt;
+import org.apache.flink.agents.api.resource.PythonResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceName;
import org.apache.flink.agents.api.resource.ResourceType;
@@ -91,11 +92,29 @@ class YamlLoaderBuildersTest {
assertThat(tool.getFunc()).isInstanceOf(PythonFunction.class);
}
+ @Test
+ void descriptorPythonMCPServerResolvesAlias() throws Exception {
+ DescriptorSpec spec =
+ M.readValue(
+ "name: x\n"
+ + "clazz: mcp\n"
+ + "type: python\n"
+ + "endpoint: http://127.0.0.1:8000/mcp\n";,
+ DescriptorSpec.class);
+ ResourceDescriptor descriptor = YamlLoader.buildDescriptor(spec,
ResourceType.MCP_SERVER);
+
+ assertThat(descriptor).isInstanceOf(PythonResourceDescriptor.class);
+
assertThat(descriptor.getModule()).isEqualTo("flink_agents.integrations.mcp.mcp");
+ assertThat(descriptor.getClazz()).isEqualTo("MCPServer");
+ assertThat(descriptor.getInitialArguments())
+ .containsExactlyEntriesOf(Map.of("endpoint",
"http://127.0.0.1:8000/mcp";));
+ }
+
@Test
void descriptorPythonRejectedForUnsupportedKind() throws Exception {
DescriptorSpec spec =
M.readValue("name: x\nclazz: anything\ntype: python\n",
DescriptorSpec.class);
- assertThatThrownBy(() -> YamlLoader.buildDescriptor(spec,
ResourceType.MCP_SERVER))
+ assertThatThrownBy(() -> YamlLoader.buildDescriptor(spec,
ResourceType.AGENT))
.hasMessageContaining("python");
}
diff --git a/docs/content/docs/development/create-agent/yaml.md
b/docs/content/docs/development/create-agent/yaml.md
index 8cb084e12..3a08e2517 100644
--- a/docs/content/docs/development/create-agent/yaml.md
+++ b/docs/content/docs/development/create-agent/yaml.md
@@ -749,6 +749,12 @@ Vector-store aliases:
| `s3_vectors` | — | S3 Vectors |
| `milvus` | — | Milvus |
+MCP-server aliases:
+
+| Alias | `type: python` | `type: java` |
+| ----- | -------------- | ------------ |
+| `mcp` | MCP Server | — |
+
The full alias tables live in `flink_agents.api.yaml.aliases` (Python) and
`org.apache.flink.agents.api.yaml.Aliases` (Java).
### Cross-language agents
@@ -814,8 +820,8 @@ Application payloads must use the supported
Cross-language support varies by component:
- `actions`, `tools`, `chat_model_connections`, `chat_model_setups`,
`embedding_model_connections`, `embedding_model_setups`, and `vector_stores`
support both directions: `type: python` when loaded from Java, and `type: java`
when loaded from Python.
-- `decision_model_connections` and `decision_model_setups` support only
Java-to-Python cross-language use: Java-loaded YAML can use Python
implementations, but Python-loaded YAML cannot use Java implementations.
-- Other components, including `mcp_servers` and external `subagents`, do not
support cross-language use.
+- `decision_model_connections`, `decision_model_setups`, and `mcp_servers`
support only Java-to-Python cross-language use: Java-loaded YAML can use Python
implementations, but Python-loaded YAML cannot use Java implementations.
+- Other components, including external `subagents`, do not support
cross-language use.
## YAML API Specification
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
b/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
index 28d870396..4dc831571 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
@@ -661,12 +661,18 @@ public class AgentPlan implements Serializable {
}
}
} else if (type == MCP_SERVER) {
- if (!entry.getValue().isEmpty()) {
- throw new UnsupportedOperationException(
- "Adding an MCP server via Agent.addResource is not
supported."
- + " Declare the MCP server with a
@MCPServer-annotated static"
- + " method on your Agent class so its
tools and prompts can be"
- + " discovered.");
+ for (Map.Entry<String, Object> kv :
entry.getValue().entrySet()) {
+ String name = kv.getKey();
+ ResourceDescriptor descriptor =
+ requireResourceDescriptor(name, type,
kv.getValue());
+ if (!isPythonResource(descriptor)) {
+ throw new UnsupportedOperationException(
+ "Adding a Java MCP server via
Agent.addResource is not supported."
+ + " Declare it with a
@MCPServer-annotated static method on"
+ + " your Agent class so its tools and
prompts can be"
+ + " discovered.");
+ }
+ addResourceProvider(new PythonResourceProvider(name,
MCP_SERVER, descriptor));
}
} else if (type == ResourceType.AGENT) {
for (Map.Entry<String, Object> kv :
entry.getValue().entrySet()) {
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProvider.java
b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProvider.java
index bfed06d2c..e2e267ed6 100644
---
a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProvider.java
+++
b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProvider.java
@@ -48,8 +48,6 @@ import static org.apache.flink.util.Preconditions.checkState;
* class, and initialization arguments.
*/
public class PythonResourceProvider extends ResourceProvider {
- private static final String MCP_MODULE =
"flink_agents.integrations.mcp.mcp";
- private static final String MCP_CLASS = "MCPServer";
private final ResourceDescriptor descriptor;
private static final Map<ResourceType, Class<?>> RESOURCE_TYPE_TO_CLASS =
@@ -96,11 +94,6 @@ public class PythonResourceProvider extends ResourceProvider
{
String pyModule = descriptor.getModule();
String pyClazz = descriptor.getClazz();
- if (getType() == ResourceType.MCP_SERVER) {
- pyModule = MCP_MODULE;
- pyClazz = MCP_CLASS;
- }
-
PyObject pyResource =
pythonResourceAdapter.initPythonResource(pyModule, pyClazz, kwargs);
try {
Constructor<?> constructor =
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
index 8f3ea4e8b..27b3eba15 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
@@ -665,6 +665,32 @@ public class AgentPlanTest {
assertThat(provider).isInstanceOf(PythonResourceProvider.class);
}
+ @Test
+ public void testYamlPythonMCPServerUsesPythonResourceProvider(@TempDir
Path tmp)
+ throws Exception {
+ Path file = tmp.resolve("mcp.yaml");
+ Files.writeString(
+ file,
+ "agents:\n"
+ + " - name: a\n"
+ + " mcp_servers:\n"
+ + " - name: yamlMcpServer\n"
+ + " clazz: mcp\n"
+ + " type: python\n"
+ + " endpoint: http://127.0.0.1:8000/mcp\n";);
+ Agent agent = YamlLoader.buildAgents(file).getAgents().get("a");
+
+ AgentPlan plan = new AgentPlan(agent);
+
+ ResourceProvider provider =
+
plan.getResourceProviders().get(ResourceType.MCP_SERVER).get("yamlMcpServer");
+ assertThat(provider).isInstanceOf(PythonResourceProvider.class);
+ assertThat(((PythonResourceProvider)
provider).getDescriptor().getModule())
+ .isEqualTo("flink_agents.integrations.mcp.mcp");
+ assertThat(((PythonResourceProvider)
provider).getDescriptor().getClazz())
+ .isEqualTo("MCPServer");
+ }
+
@Test
public void testAddResourceMCPServerRejectedWithGuidance() {
Agent agent = new Agent();
diff --git
a/plan/src/test/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProviderTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProviderTest.java
index b074083ad..bb9150c5a 100644
---
a/plan/src/test/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProviderTest.java
+++
b/plan/src/test/java/org/apache/flink/agents/plan/resourceprovider/PythonResourceProviderTest.java
@@ -69,6 +69,28 @@ class PythonResourceProviderTest {
verify(pythonResource).close();
}
+ @Test
+ void mcpServerUsesImplementationFromDescriptor() throws Exception {
+ PythonResourceAdapter adapter = mock(PythonResourceAdapter.class);
+ PyObject pythonResource = mock(PyObject.class);
+ ResourceDescriptor descriptor =
+ new PythonResourceDescriptor(
+ "custom.mcp", "CustomMCPServer", Map.of("endpoint",
"http://mcp";));
+ PythonResourceProvider provider =
+ new PythonResourceProvider("server", ResourceType.MCP_SERVER,
descriptor);
+ provider.setPythonResourceAdapter(adapter);
+ when(adapter.initPythonResource(
+ "custom.mcp", "CustomMCPServer", Map.of("endpoint",
"http://mcp";)))
+ .thenReturn(pythonResource);
+
+ Resource resource = provider.provide(mock(ResourceContext.class));
+
+ verify(adapter)
+ .initPythonResource(
+ "custom.mcp", "CustomMCPServer", Map.of("endpoint",
"http://mcp";));
+ resource.close();
+ }
+
@Test
void closesPythonHandleWhenWrapperConstructionFails() throws Exception {
PythonResourceAdapter adapter = mock(PythonResourceAdapter.class);
diff --git a/python/flink_agents/api/yaml/aliases.py
b/python/flink_agents/api/yaml/aliases.py
index 076202a5e..20f10b571 100644
--- a/python/flink_agents/api/yaml/aliases.py
+++ b/python/flink_agents/api/yaml/aliases.py
@@ -157,6 +157,11 @@ CLAZZ_ALIASES: Dict[ResourceType, Dict[str, Dict[str,
str]]] = {
"milvus": ResourceName.VectorStore.Java.MILVUS_VECTOR_STORE,
},
},
+ ResourceType.MCP_SERVER: {
+ "python": {
+ "mcp": ResourceName.MCP_SERVER,
+ },
+ },
}
diff --git a/python/flink_agents/api/yaml/tests/test_aliases.py
b/python/flink_agents/api/yaml/tests/test_aliases.py
index 11ecda7d9..9b26fd0a5 100644
--- a/python/flink_agents/api/yaml/tests/test_aliases.py
+++ b/python/flink_agents/api/yaml/tests/test_aliases.py
@@ -94,6 +94,13 @@ def test_resolve_clazz_replaces_alias_per_resource_type() ->
None:
assert embed_conn.endswith("OllamaEmbeddingModelConnection")
+def test_resolve_clazz_replaces_mcp_alias() -> None:
+ assert (
+ resolve_clazz("mcp", ResourceType.MCP_SERVER)
+ == "flink_agents.integrations.mcp.mcp.MCPServer"
+ )
+
+
def test_resolve_clazz_passes_through_fqn() -> None:
assert (
resolve_clazz("my.custom.Klass", ResourceType.CHAT_MODEL) ==
"my.custom.Klass"
diff --git a/python/flink_agents/api/yaml/tests/test_loader.py
b/python/flink_agents/api/yaml/tests/test_loader.py
index 3bd0c712d..bd1052c94 100644
--- a/python/flink_agents/api/yaml/tests/test_loader.py
+++ b/python/flink_agents/api/yaml/tests/test_loader.py
@@ -642,6 +642,34 @@ def test_build_agents_supports_type_java(tmp_path: Path)
-> None:
assert conn.arguments["requestTimeout"] == 120
+def test_build_agents_resolves_mcp_alias_at_both_scopes(tmp_path: Path) ->
None:
+ source = tmp_path / "mcp.yaml"
+ source.write_text(
+ """mcp_servers:
+ - name: shared_mcp
+ clazz: mcp
+ endpoint: http://127.0.0.1:8000/mcp
+agents:
+ - name: a
+ mcp_servers:
+ - name: local_mcp
+ clazz: mcp
+ type: python
+ endpoint: http://127.0.0.1:8001/mcp
+"""
+ )
+
+ agents, shared_resources, _ = build_agents(source)
+
+ for descriptor in [
+ shared_resources[ResourceType.MCP_SERVER]["shared_mcp"],
+ agents["a"].resources[ResourceType.MCP_SERVER]["local_mcp"],
+ ]:
+ assert descriptor.language == "python"
+ assert descriptor.target_module == "flink_agents.integrations.mcp.mcp"
+ assert descriptor.target_clazz == "MCPServer"
+
+
def test_build_agents_rejects_type_java_for_unsupported_resource(
tmp_path: Path,
) -> None:
