feat: add agentic_generative_ui agent

Author: brandonmcconnell

integrations/adk-middleware/python/examples/server/__init__.py

@@ -19,6 +19,7 @@
 
 from .api import (
     agentic_chat_app,
+    agentic_generative_ui_app,
     tool_based_generative_ui_app,
     human_in_the_loop_app,
     shared_state_app,
@@ -30,6 +31,7 @@
 
 # Include routers instead of mounting apps to show routes in docs
 app.include_router(agentic_chat_app.router, prefix='/chat', tags=['Agentic Chat'])
+app.include_router(agentic_generative_ui_app.router, prefix='/adk-agentic-generative-ui', tags=['Agentic Generative UI'])
 app.include_router(tool_based_generative_ui_app.router, prefix='/adk-tool-based-generative-ui', tags=['Tool Based Generative UI'])
 app.include_router(human_in_the_loop_app.router, prefix='/adk-human-in-loop-agent', tags=['Human in the Loop'])
 app.include_router(shared_state_app.router, prefix='/adk-shared-state-agent', tags=['Shared State'])
@@ -43,6 +45,7 @@ async def root():
         "message": "ADK Middleware is running!",
         "endpoints": {
             "chat": "/chat",
+            "agentic_generative_ui": "/adk-agentic-generative-ui",
             "tool_based_generative_ui": "/adk-tool-based-generative-ui",
             "human_in_the_loop": "/adk-human-in-loop-agent",
             "shared_state": "/adk-shared-state-agent",
@@ -83,6 +86,7 @@ def main():
     print("Starting ADK Middleware server...")
     print(f"Available endpoints:")
     print(f"  • Chat: http://localhost:{port}/chat")
+    print(f"  • Agentic Generative UI: http://localhost:{port}/adk-agentic-generative-ui")
     print(f"  • Tool Based Generative UI: http://localhost:{port}/adk-tool-based-generative-ui")
     print(f"  • Human in the Loop: http://localhost:{port}/adk-human-in-loop-agent")
     print(f"  • Shared State: http://localhost:{port}/adk-shared-state-agent")

integrations/adk-middleware/python/examples/server/api/__init__.py

@@ -1,6 +1,7 @@
 """API modules for ADK middleware examples."""
 
 from .agentic_chat import app as agentic_chat_app
+from .agentic_generative_ui import app as agentic_generative_ui_app
 from .tool_based_generative_ui import app as tool_based_generative_ui_app
 from .human_in_the_loop import app as human_in_the_loop_app
 from .shared_state import app as shared_state_app
@@ -9,6 +10,7 @@
 
 __all__ = [
     "agentic_chat_app",
+    "agentic_generative_ui_app",
     "tool_based_generative_ui_app",
     "human_in_the_loop_app",
     "shared_state_app",

integrations/adk-middleware/python/examples/server/api/agentic_generative_ui.py

@@ -0,0 +1,135 @@
+"""Agentic Generative UI feature."""
+
+from __future__ import annotations
+
+from textwrap import dedent
+from typing import Any, Literal
+
+from fastapi import FastAPI
+from pydantic import BaseModel, Field
+
+from ag_ui.core import EventType, StateDeltaEvent, StateSnapshotEvent
+from ag_ui_adk import ADKAgent, add_adk_fastapi_endpoint
+from google.adk.agents import LlmAgent
+
+StepStatus = Literal['pending', 'completed']
+
+
+class Step(BaseModel):
+    """Represents a step in a plan."""
+
+    description: str = Field(description='The description of the step')
+    status: StepStatus = Field(
+        default='pending',
+        description='The status of the step (e.g., pending, completed)',
+    )
+
+
+class Plan(BaseModel):
+    """Represents a plan with multiple steps."""
+
+    steps: list[Step] = Field(default_factory=list, description='The steps in the plan')
+
+
+class JSONPatchOp(BaseModel):
+    """A class representing a JSON Patch operation (RFC 6902)."""
+
+    op: Literal['add', 'remove', 'replace', 'move', 'copy', 'test'] = Field(
+        description='The operation to perform: add, remove, replace, move, copy, or test',
+    )
+    path: str = Field(description='JSON Pointer (RFC 6901) to the target location')
+    value: Any = Field(
+        default=None,
+        description='The value to apply (for add, replace operations)',
+    )
+    from_: str | None = Field(
+        default=None,
+        alias='from',
+        description='Source path (for move, copy operations)',
+    )
+
+
+async def create_plan(steps: list[str]) -> StateSnapshotEvent:
+    """Create a plan with multiple steps.
+
+    Args:
+        steps (list[str]): List of step descriptions to create the plan.
+
+    Returns:
+        StateSnapshotEvent: Event containing the initial state of the steps.
+    """
+    plan: Plan = Plan(
+        steps=[Step(description=step) for step in steps],
+    )
+    return StateSnapshotEvent(
+        type=EventType.STATE_SNAPSHOT,
+        snapshot=plan.model_dump(),
+    )
+
+
+async def update_plan_step(
+    index: int, description: str | None = None, status: StepStatus | None = None
+) -> StateDeltaEvent:
+    """Update the plan with new steps or changes.
+
+    Args:
+        index (int): The index of the step to update.
+        description (str | None): The new description for the step.
+        status (StepStatus | None): The new status for the step.
+
+    Returns:
+        StateDeltaEvent: Event containing the changes made to the plan.
+    """
+    changes: list[JSONPatchOp] = []
+    if description is not None:
+        changes.append(
+            JSONPatchOp(
+                op='replace', path=f'/steps/{index}/description', value=description
+            )
+        )
+    if status is not None:
+        changes.append(
+            JSONPatchOp(op='replace', path=f'/steps/{index}/status', value=status)
+        )
+    return StateDeltaEvent(
+        type=EventType.STATE_DELTA,
+        delta=changes,
+    )
+
+
+# Create the ADK agent
+agent = LlmAgent(
+    name="planner",
+    model="gemini-2.0-flash",
+    instruction=dedent(
+        """
+        When planning use tools only, without any other messages.
+        IMPORTANT:
+        - Use the `create_plan` tool to set the initial state of the steps
+        - Use the `update_plan_step` tool to update the status of each step
+        - Do NOT repeat the plan or summarise it in a message
+        - Do NOT confirm the creation or updates in a message
+        - Do NOT ask the user for additional information or next steps
+        - Do NOT leave a plan hanging, always complete the plan via `update_plan_step` if one is ongoing.
+
+        Only one plan can be active at a time, so do not call the `create_plan` tool
+        again until all the steps in current plan are completed.
+        """
+    ),
+    tools=[create_plan, update_plan_step],
+)
+
+# Create ADK middleware agent instance
+adk_agent = ADKAgent(
+    adk_agent=agent,
+    app_name="demo_app",
+    user_id="demo_user",
+    session_timeout_seconds=3600,
+    use_in_memory_services=True,
+)
+
+# Create FastAPI app
+app = FastAPI(title="ADK Middleware Agentic Generative UI")
+
+# Add the ADK endpoint
+add_adk_fastapi_endpoint(app, adk_agent, path="/")