How to Make Agents Talk to Each Other (and Your App) Using A2A + AG-UI

npx copilotkit@latest create -f a2a

npm install

cd agents
python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install -r requirements.txt
cd ..

cp .env.example .env
# Edit .env and add your API keys:
# GOOGLE_API_KEY=your_google_api_key
# OPENAI_API_KEY=your_openai_api_key

npm run dev

cd agents

python -m venv .venv

source .venv/bin/activate  # On Windows: .venv\Scripts\activate

pip install -r requirements.txt

# Import Google ADK components for LLM agent creation
from google.adk.agents import LlmAgent

# === ORCHESTRATOR AGENT CONFIGURATION ===
# Create the main orchestrator agent using Google ADK's LlmAgent
# This agent coordinates all travel planning activities and manages the workflow

orchestrator_agent = LlmAgent(
    name="OrchestratorAgent",
    model="gemini-2.5-pro",  # Use the more powerful Pro model for complex orchestration
    instruction="""
    You are a travel planning orchestrator agent. Your role is to coordinate specialized agents
    to create personalized travel plans.

    AVAILABLE SPECIALIZED AGENTS:

    1. **Itinerary Agent** (LangGraph) - Creates day-by-day travel itineraries with activities
    2. **Restaurant Agent** (LangGraph) - Recommends restaurants for breakfast, lunch, and dinner by day
    3. **Weather Agent** (ADK) - Provides weather forecasts and packing advice
    4. **Budget Agent** (ADK) - Estimates travel costs and creates budget breakdowns

    CRITICAL CONSTRAINTS:
    - You MUST call agents ONE AT A TIME, never make multiple tool calls simultaneously
    - After making a tool call, WAIT for the result before making another tool call
    - Do NOT make parallel/concurrent tool calls - this is not supported

    RECOMMENDED WORKFLOW FOR TRAVEL PLANNING:

    // ...

    """,
)

# Import AG-UI ADK components for frontend integration
from ag_ui_adk import ADKAgent

// ...

# === AG-UI PROTOCOL INTEGRATION ===
# Wrap the orchestrator agent with AG-UI Protocol capabilities
# This enables frontend communication and provides the interface for user interactions

adk_orchestrator_agent = ADKAgent(
    adk_agent=orchestrator_agent,          # The core LLM agent we created above
    app_name="orchestrator_app",           # Unique application identifier
    user_id="demo_user",                   # Default user ID for demo purposes
    session_timeout_seconds=3600,          # Session timeout (1 hour)
    use_in_memory_services=True            # Use in-memory storage for simplicity
)

# Import necessary libraries for web server and environment variables
import os
import uvicorn

# Import FastAPI for creating HTTP endpoints
from fastapi import FastAPI

// ...

# === FASTAPI WEB APPLICATION SETUP ===
# Create the FastAPI application that will serve the orchestrator agent
# This provides HTTP endpoints for the AG-UI Protocol communication

app = FastAPI(title="Travel Planning Orchestrator (ADK)")

# Add the ADK agent endpoint to the FastAPI application
# This creates the necessary routes for AG-UI Protocol communication
add_adk_fastapi_endpoint(app, adk_orchestrator_agent, path="/")

# === MAIN APPLICATION ENTRY POINT ===
if __name__ == "__main__":
    """
    Main entry point when the script is run directly.

    This function:
    1. Checks for required environment variables (API keys)
    2. Configures the server port
    3. Starts the uvicorn server with the FastAPI application
    """

    # Check for required Google API key
    if not os.getenv("GOOGLE_API_KEY"):
        print("⚠️  Warning: GOOGLE_API_KEY environment variable not set!")
        print("   Set it with: export GOOGLE_API_KEY='your-key-here'")
        print("   Get a key from: https://aistudio.google.com/app/apikey")
        print()

    # Get server port from environment variable, default to 9000
    port = int(os.getenv("ORCHESTRATOR_PORT", 9000))

    # Start the server with detailed information
    print(f"🚀 Starting Orchestrator Agent (ADK + AG-UI) on http://localhost:{port}")

    # Run the FastAPI application using uvicorn
    # host="0.0.0.0" allows external connections
    # port is configurable via the environment variable
    uvicorn.run(app, host="0.0.0.0", port=port)

# Import LangGraph components for workflow management
from langgraph.graph import StateGraph, END

// ...

# === MAIN AGENT CLASS ===
class ItineraryAgent:
    """
    Main agent class that handles itinerary generation using LangGraph workflow.

    """

    def __init__(self):

        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

        # Build and compile the LangGraph workflow
        self.graph = self._build_graph()

    def _build_graph(self):
        workflow = StateGraph(ItineraryState)
        workflow.add_node("parse_request", self._parse_request)
        workflow.add_node("create_itinerary", self._create_itinerary)
        workflow.set_entry_point("parse_request")
        workflow.add_edge("parse_request", "create_itinerary")
        workflow.add_edge("create_itinerary", END)

        # Compile the workflow into an executable graph
        return workflow.compile()

    def _parse_request(self, state: ItineraryState) -> ItineraryState:
        message = state["message"]

        # Create a focused prompt for the extraction task
        prompt = f"""
        Extract the destination and number of days from this travel request.
        Return ONLY a JSON string with 'destination' and 'days' fields.

        Request: {message}

        Example output: {{"destination": "Tokyo", "days": 3}}
        """

        # Get LLM response for parsing
        response = self.llm.invoke(prompt)

        # Debug: Print the LLM response for troubleshooting
        print(response.content)

        try:
            # Attempt to parse the JSON response
            parsed = json.loads(response.content)
            state["destination"] = parsed.get("destination", "Unknown")
            state["days"] = int(parsed.get("days", 3))
        except:
            # Fallback values if parsing fails
            print("⚠️  Failed to parse request, using defaults")
            state["destination"] = "Unknown"
            state["days"] = 3

        return state

    def _create_itinerary(self, state: ItineraryState) -> ItineraryState:
        destination = state["destination"]
        days = state["days"]

        # Create detailed prompt for itinerary generation
        prompt = f"""
        Create a detailed {days}-day travel itinerary for {destination}.

        // ...

        Make it realistic, interesting, and include specific place names.
        Return ONLY valid JSON, no markdown, no other text.
        """

        # Generate itinerary using LLM
        response = self.llm.invoke(prompt)
        content = response.content.strip()

        # Clean up response - remove markdown formatting if present
        if "```

json" in content:
            content = content.split("

```json")[1].split("```

")[0].strip()
        elif "

```" in content:
            content = content.split("```

")[1].split("

```")[0].strip()

        try:
            # Step 1: Parse JSON from LLM response
            structured_data = json.loads(content)

            # Step 2: Validate structure using Pydantic model
            validated_itinerary = StructuredItinerary(**structured_data)

            # Step 3: Store both validated data and formatted JSON string
            state["structured_itinerary"] = validated_itinerary.model_dump()
            state["itinerary"] = json.dumps(validated_itinerary.model_dump(), indent=2)

            print("✅ Successfully created structured itinerary")

        // ...

        return state

    async def invoke(self, message: Message) -> str:

        # Extract text content from A2A message format
        message_text = message.parts[0].root.text
        print("Invoking itinerary agent with message: ", message_text)

        # Execute the LangGraph workflow with initial state
        result = self.graph.invoke({
            "message"