feat(vefaas): add mcp server (#51)

* test(vefaas): update Unitest of CloudAgentEngine and CloudApp

* fix(test): test_cloud typer dependency

* test(vefaas): volcenginesdkcore dependency, uv sync --all-extras

* test(vefaas): modify chinese annotation

* test(vefaas): fix deploy docs

* feat(vefaas): add mcp server pre-commit

* feat(vefaas): add mcp server pyproject cli

* fix(vefaas): fix mcp mount a2a

* fix(vefaas): fix run.sh

* fix(vefaas): fix app.py pre

* fix(vefaas): VeA2AServer add post run_agent to mcp server

* fix(vefaas): mcp server in app.py

---------

Co-authored-by: tangou <[email protected]>
Co-authored-by: hanzhi <[email protected]>

Author: floritange

pyproject.toml

@@ -43,7 +43,8 @@ cli = [
     "volcengine-python-sdk==4.0.3", # For Volcengine API
     "typer>=0.16.0",                # For command-line implementation
     "streamlit==1.46.1",            # For running VeADK studio app
-    "agent-pilot-sdk>=0.0.9"        # Prompt optimization by Volcengine AgentPilot/PromptPilot toolkits
+    "agent-pilot-sdk>=0.0.9",       # Prompt optimization by Volcengine AgentPilot/PromptPilot toolkits
+    "fastmcp>=2.11.3",              # For running MCP
 ]
 dev = [
     "pre-commit>=4.2.0",            # Format checking

veadk/a2a/ve_a2a_server.py

@@ -48,8 +48,6 @@ def build(self) -> FastAPI:
         )
         app = app_application.build()  # build routes
 
-        # import uvicorn
-        # uvicorn.run(app, host="127.0.0.1", port=8000)
         return app
 
 

veadk/cli/main.py

@@ -98,14 +98,14 @@ def init():
     )
 
     deploy_mode_options = {
-        "1": "A2A Server",
+        "1": "A2A/MCP Server",
         "2": "VeADK Studio",
         "3": "VeADK Web / Google ADK Web",
     }
 
     deploy_mode = Prompt.ask(
         """Choose your deploy mode:
-1. A2A Server
+1. A2A/MCP Server
 2. VeADK Studio
 3. VeADK Web / Google ADK Web
 """,

veadk/cli/services/vefaas/template/deploy.py

@@ -15,7 +15,9 @@
 import asyncio
 from pathlib import Path
 
+
 from veadk.cloud.cloud_agent_engine import CloudAgentEngine
+from fastmcp.client import Client
 
 SESSION_ID = "cloud_app_test_session"
 USER_ID = "cloud_app_test_user"
@@ -30,7 +32,6 @@
 
 async def main():
     engine = CloudAgentEngine()
-
     cloud_app = engine.deploy(
         path=str(Path(__file__).parent / "src"),
         application_name=VEFAAS_APPLICATION_NAME,
@@ -40,16 +41,42 @@ async def main():
         use_studio=USE_STUDIO,
         use_adk_web=USE_ADK_WEB,
     )
-
-    if not USE_STUDIO and not USE_ADK_WEB:
+    query_example = "How is the weather like in Beijing?"
+    if not USE_STUDIO and (not USE_ADK_WEB):
+        print("### A2A example ###")
         response_message = await cloud_app.message_send(
-            "How is the weather like in Beijing?", SESSION_ID, USER_ID
+            query_example, SESSION_ID, USER_ID
         )
         print(f"VeFaaS application ID: {cloud_app.vefaas_application_id}")
         print(f"Message ID: {response_message.messageId}")
         print(
             f"Response from {cloud_app.vefaas_endpoint}: {response_message.parts[0].root.text}"
         )
+
+        print("### MCP example ###")
+        # cloud_app = CloudApp(vefaas_application_name=VEFAAS_APPLICATION_NAME)
+        endpoint = cloud_app._get_vefaas_endpoint()
+        print(f"endpoint:{endpoint}")
+        # Connect to MCP server
+        client = Client(f"{endpoint}/mcp")
+
+        async with client:
+            # List available tools
+            tools = await client.list_tools()
+            print(f"tools: {tools}\n")
+
+            # Call run_agent tool, pass user input and session information
+            res = await client.call_tool(
+                "run_agent",
+                {
+                    "user_input": query_example,
+                    "session_id": SESSION_ID,
+                    "user_id": USER_ID,
+                },
+            )
+            print(f"VeFaaS application ID: {cloud_app.vefaas_application_id}")
+            print(f"Response from {cloud_app.vefaas_endpoint}: {res}")
+
     else:
         print(f"Web is running at: {cloud_app.vefaas_endpoint}")
 

veadk/cli/services/vefaas/template/src/app.py

@@ -13,11 +13,14 @@
 # limitations under the License.
 
 import os
-
 from agent import agent, app_name, short_term_memory
 from veadk.a2a.ve_a2a_server import init_app
 from veadk.tracing.base_tracer import BaseTracer
 from veadk.tracing.telemetry.opentelemetry_tracer import OpentelemetryTracer
+from veadk.runner import Runner
+from contextlib import asynccontextmanager
+from fastmcp import FastMCP
+from fastapi import FastAPI
 
 
 # ==============================================================================
@@ -62,9 +65,69 @@
 # Tracer Config ================================================================
 # ==============================================================================
 
-app = init_app(
-    server_url="0.0.0.0",  # Automatic identification is not supported yet.
+# Create A2A app
+a2a_app = init_app(
+    server_url="0.0.0.0",
     app_name=app_name,
     agent=agent,
     short_term_memory=short_term_memory,
 )
+
+# Add a2a app to fastmcp
+runner = Runner(
+    agent=agent,
+    short_term_memory=short_term_memory,
+    app_name=app_name,
+    user_id="",
+)
+
+
+# mcp server
+@a2a_app.post("/run_agent", operation_id="run_agent", tags=["mcp"])
+async def run_agent(
+    user_input: str,
+    user_id: str = "unknown_user",
+    session_id: str = "unknown_session",
+) -> str:
+    """
+    Execute agent with user input and return final output
+    Args:
+        user_input: User's input message
+        user_id: User identifier
+        session_id: Session identifier
+    Returns:
+        Final agent response
+    """
+    # Set user_id for runner
+    runner.user_id = user_id
+
+    # Running agent and get final output
+    final_output = await runner.run(
+        messages=user_input,
+        session_id=session_id,
+    )
+    return final_output
+
+
+mcp = FastMCP.from_fastapi(app=a2a_app, name=app_name, include_tags={"mcp"})
+
+# Create MCP ASGI app
+mcp_app = mcp.http_app(path="/")
+
+
+# Combined lifespan management
+@asynccontextmanager
+async def combined_lifespan(app: FastAPI):
+    async with mcp_app.lifespan(app):
+        yield
+
+
+# Create main FastAPI app with combined lifespan
+app = FastAPI(title=a2a_app.title, version=a2a_app.version, lifespan=combined_lifespan)
+
+# Mount A2A routes to main app
+for route in a2a_app.routes:
+    app.routes.append(route)
+
+# Mount MCP server at /mcp endpoint
+app.mount("/mcp", mcp_app)

veadk/cli/services/vefaas/template/src/requirements.txt

@@ -3,4 +3,5 @@ opensearch-py==2.8.0
 agent-pilot-sdk>=0.0.9 # extra dep for prompt optimization in veadk studio
 typer>=0.16.0
 uvicorn[standard]
-fastapi
+fastapi
+fastmcp

veadk/cli/services/vefaas/template/src/run.sh

@@ -38,6 +38,8 @@ python3 -m pip install uvicorn[standard]
 
 python3 -m pip install fastapi
 
+python3 -m pip install fastmcp
+
 USE_STUDIO=${USE_STUDIO:-False}
 USE_ADK_WEB=${USE_ADK_WEB:-False}