Mcp Apps 2 : Introducing MCP Apps to A2UI Agent

agent_sdks/python/src/a2ui/adk/a2a_extension/send_a2ui_to_client_toolset.py

@@ -339,8 +339,9 @@ class A2uiPartConverter:
   catalog to validate and fix JSON payloads.
   """
 
-  def __init__(self, a2ui_catalog: A2uiCatalog):
+  def __init__(self, a2ui_catalog: A2uiCatalog, bypass_tool_check: bool = False):
     self._catalog = a2ui_catalog
+    self._bypass_tool_check = bypass_tool_check
 
   def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
     """Converts a GenAI part to A2A parts, with A2UI validation.
@@ -352,30 +353,39 @@ def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
         A list of A2A parts.
     """
     # 1. Handle Tool Responses (FunctionResponse)
-    if (
-        (function_response := part.function_response)
-        and function_response.name
-        == SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_NAME
-    ):
-      if (
-          SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_ERROR_KEY
-          in function_response.response
-      ):
-        logger.warning(
-            "A2UI tool call failed:"
-            f" {function_response.response[SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_ERROR_KEY]}"
-        )
-        return []
-
-      # The tool returns the list of messages directly on success
-      json_data = function_response.response.get(
-          SendA2uiToClientToolset._SendA2uiJsonToClientTool.VALIDATED_A2UI_JSON_KEY
+    if function_response := part.function_response:
+      is_send_a2ui_json_to_client_response = (
+          function_response.name
+          == SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_NAME
       )
-      if not json_data:
-        logger.info("No result in A2UI tool response")
-        return []
 
-      return [create_a2ui_part(message) for message in json_data]
+      if is_send_a2ui_json_to_client_response or self._bypass_tool_check:
+        response_dict = function_response.response or {}
+
+        if (
+            SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_ERROR_KEY
+            in response_dict
+        ):
+          logger.warning(
+              "A2UI tool call failed:"
+              f" {response_dict[SendA2uiToClientToolset._SendA2uiJsonToClientTool.TOOL_ERROR_KEY]}"
+          )
+          return []
+
+        if (
+            isinstance(response_dict, dict)
+            and SendA2uiToClientToolset._SendA2uiJsonToClientTool.VALIDATED_A2UI_JSON_KEY
+            in response_dict
+        ):
+          json_data = response_dict.get(
+              SendA2uiToClientToolset._SendA2uiJsonToClientTool.VALIDATED_A2UI_JSON_KEY
+          )
+          if json_data:
+            return [create_a2ui_part(message) for message in json_data]
+
+        if is_send_a2ui_json_to_client_response:
+          logger.info("No result in A2UI tool response")
+          return []
 
     # 2. Handle Tool Calls (FunctionCall) - Skip sending to client
     if (
@@ -403,8 +413,11 @@ class A2uiEventConverter:
   catalog is session-specific.
   """
 
-  def __init__(self, catalog_key: str = "system:a2ui_catalog"):
+  def __init__(
+      self, catalog_key: str = "system:a2ui_catalog", bypass_tool_check: bool = False
+  ):
     self._catalog_key = catalog_key
+    self._bypass_tool_check = bypass_tool_check
 
   def __call__(
       self,
@@ -420,7 +433,9 @@ def __call__(
     catalog = invocation_context.session.state.get(self._catalog_key)
     if catalog:
       # Use the catalog-aware part converter
-      effective_converter = A2uiPartConverter(catalog).convert
+      effective_converter = A2uiPartConverter(
+          catalog, bypass_tool_check=self._bypass_tool_check
+      ).convert
     else:
       effective_converter = part_converter_func
 

samples/agent/adk/mcp_app_proxy/.env.example

@@ -0,0 +1,8 @@
+# Copy this file to .env and fill in your API key
+# Get your API key at: https://aistudio.google.com/apikey
+
+GEMINI_API_KEY=your_gemini_api_key_here
+
+# Change the following values to point to your own MCP Server.
+MCP_SERVER_HOST=localhost
+MCP_SERVER_PORT=8000  

samples/agent/adk/mcp_app_proxy/README.md

@@ -0,0 +1,65 @@
+# A2UI MCP Apps Proxy Agent sample
+
+This sample uses the Agent Development Kit (ADK) along with the A2A protocol to
+create a simple "MCP Apps Proxy" agent that is hosted as an A2A server.
+
+## Prerequisites
+
+- Python 3.9 or higher
+- [UV](https://docs.astral.sh/uv/)
+- Access to an LLM and API Key
+
+## Running the sample
+
+1. Run the MCP Server that serves the MCP Apps. ([Link to
+   instructions](../../agent/mcp/README.md))
+
+
+2. Navigate to the samples directory:
+
+    ```bash
+    cd samples/agent/adk/mcp_app_proxy
+    ```
+
+2. Create an environment file with your API key:
+
+   ```bash
+   cp .env.example .env
+   # Edit .env with your actual API key (do not commit .env)
+   ```
+
+3. Run the server:
+
+   ```bash
+   uv run .
+   ```
+
+
+## Disclaimer
+
+Important: The sample code provided is for demonstration purposes and
+illustrates the mechanics of A2UI and the Agent-to-Agent (A2A) protocol. When
+building production applications, it is critical to treat any agent operating
+outside of your direct control as a potentially untrusted entity.
+
+All operational data received from an external agent—including its AgentCard,
+messages, artifacts, and task statuses—should be handled as untrusted input. For
+example, a malicious agent could provide crafted data in its fields (e.g., name,
+skills.description) that, if used without sanitization to construct prompts for
+a Large Language Model (LLM), could expose your application to prompt injection
+attacks.
+
+Similarly, any UI definition or data stream received must be treated as
+untrusted. Malicious agents could attempt to spoof legitimate interfaces to
+deceive users (phishing), inject malicious scripts via property values (XSS), or
+generate excessive layout complexity to degrade client performance (DoS). If
+your application supports optional embedded content (such as iframes or web
+views), additional care must be taken to prevent exposure to malicious external
+sites.
+
+Developer Responsibility: Failure to properly validate data and strictly sandbox
+rendered content can introduce severe vulnerabilities. Developers are
+responsible for implementing appropriate security measures—such as input
+sanitization, Content Security Policies (CSP), strict isolation for optional
+embedded content, and secure credential handling—to protect their systems and
+users.

samples/agent/adk/mcp_app_proxy/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import agent

samples/agent/adk/mcp_app_proxy/__main__.py

@@ -0,0 +1,185 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from a2a.server.apps import A2AStarletteApplication
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2ui.core.schema.constants import VERSION_0_8
+from a2ui.core.schema.manager import A2uiSchemaManager, CatalogConfig
+from agent import McpAppProxyAgent
+from agent_executor import McpAppProxyAgentExecutor, get_a2ui_enabled, get_a2ui_catalog, get_a2ui_examples
+from dotenv import load_dotenv
+from google.adk.artifacts import InMemoryArtifactService
+from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
+from google.adk.models.lite_llm import LiteLlm
+from google.adk.runners import Runner
+from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.adk.tools.tool_context import ToolContext
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+from starlette.middleware.cors import CORSMiddleware
+import anyio
+import click
+import httpx
+import logging
+import os
+import traceback
+import urllib.parse
+
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class MissingAPIKeyError(Exception):
+  """Exception for missing API key."""
+
+
+@click.command()
+@click.option("--host", default="localhost")
+@click.option("--port", default=10006)
+def main(host, port):
+  try:
+    if not os.getenv("GOOGLE_GENAI_USE_VERTEXAI") == "TRUE":
+      if not os.getenv("GEMINI_API_KEY"):
+        raise MissingAPIKeyError(
+            "GEMINI_API_KEY environment variable not set and GOOGLE_GENAI_USE_VERTEXAI"
+            " is not TRUE."
+        )
+
+    lite_llm_model = os.getenv("LITELLM_MODEL", "gemini/gemini-2.5-flash")
+    base_url = f"http://{host}:{port}"
+
+    schema_manager = A2uiSchemaManager(
+        VERSION_0_8,
+        catalogs=[
+            CatalogConfig.from_path(
+                name="mcp_app_proxy",
+                catalog_path="mcp_app_catalog.json",
+            ),
+        ],
+        accepts_inline_catalogs=True,
+    )
+
+    # Define get_calculator_app tool in a way that the LlmAgent can use.
+    async def get_calculator_app(tool_context: ToolContext):
+      """Fetches the calculator app."""
+      # Connect to the MCP server via SSE
+      mcp_server_host = os.getenv("MCP_SERVER_HOST", "localhost")
+      mcp_server_port = os.getenv("MCP_SERVER_PORT", "8000")
+      sse_url = f"http://{mcp_server_host}:{mcp_server_port}/sse"
+
+      try:
+        async with sse_client(sse_url) as streams:
+          async with ClientSession(streams[0], streams[1]) as session:
+            await session.initialize()
+
+            # Read the resource
+            result = await session.read_resource("ui://calculator/app")
+
+            # Package the resource as an A2UI message
+            if result.contents and hasattr(result.contents[0], "text"):
+              html_content = result.contents[0].text
+              encoded_html = "url_encoded:" + urllib.parse.quote(html_content)
+              messages = [
+                  {
+                      "beginRendering": {
+                          "surfaceId": "calculator_surface",
+                          "root": "calculator_app_root",
+                      },
+                  },
+                  {
+                      "surfaceUpdate": {
+                          "surfaceId": "calculator_surface",
+                          "components": [{
+                              "id": "calculator_app_root",
+                              "component": {
+                                  "McpApp": {
+                                      "content": {"literalString": encoded_html},
+                                      "title": {"literalString": "Calculator"},
+                                  }
+                              },
+                          }],
+                      },
+                  },
+              ]
+              tool_context.actions.skip_summarization = True
+              return {"validated_a2ui_json": messages}
+            else:
+              logger.error("Failed to get text content from resource")
+              return {"error": "Could not fetch calculator app content."}
+
+      except Exception as e:
+        logger.error(f"Error fetching calculator app: {e} {traceback.format_exc()}")
+        return {"error": f"Failed to connect to MCP server or fetch app. Details: {e}"}
+
+    tools = [get_calculator_app]
+
+    agent = McpAppProxyAgent(
+        base_url=base_url,
+        model=LiteLlm(model=lite_llm_model),
+        schema_manager=schema_manager,
+        a2ui_enabled_provider=get_a2ui_enabled,
+        a2ui_catalog_provider=get_a2ui_catalog,
+        a2ui_examples_provider=get_a2ui_examples,
+        tools=tools,
+    )
+
+    runner = Runner(
+        app_name=agent.name,
+        agent=agent,
+        artifact_service=InMemoryArtifactService(),
+        session_service=InMemorySessionService(),
+        memory_service=InMemoryMemoryService(),
+    )
+
+    agent_executor = McpAppProxyAgentExecutor(
+        base_url=base_url,
+        runner=runner,
+        schema_manager=schema_manager,
+    )
+
+    request_handler = DefaultRequestHandler(
+        agent_executor=agent_executor,
+        task_store=InMemoryTaskStore(),
+    )
+    server = A2AStarletteApplication(
+        agent_card=agent.get_agent_card(), http_handler=request_handler
+    )
+    import uvicorn
+
+    app = server.build()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["http://localhost:5173"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    uvicorn.run(app, host=host, port=port)
+  except MissingAPIKeyError as e:
+    logger.error(f"Error: {e} {traceback.format_exc()}")
+    exit(1)
+  except Exception as e:
+    logger.error(
+        f"An error occurred during server startup: {e} {traceback.format_exc()}"
+    )
+    exit(1)
+
+
+if __name__ == "__main__":
+  main()