AdkApp how to use Streaming Response

[ktk94]

How to send a request to AdkApp deployed in the Vertex AI Agent Engine to receive a streaming response

AGENT_MAP = {
    "review_agent": "https://us-central1-a2a-agent-gateway.vertexai.cloud.goog/v1/projects/PROJECT_ID/locations/us-central1/agents/REVIEW_AGENT_ID",
    "preparation_agent": "https://us-central1-a2a-agent-gateway.vertexai.cloud.goog/v1/projects/PROJECT_ID/locations/us-central1/agents/preparation_agent_id",
}
def get_google_auth_token():
    try:
        credential, project_id = google.auth.default(
            scopes=['https://www.googleapis.com/auth/cloud-platform']
        )
        if credential.token is None:
            pass
        if not credential.valid:
            credentials.refresh(Request())
        token = credential.token
        if token is None:
            raise Exception("Failed to obtain Google Cloud authentication token.")
        return token
async def delegate_agent(
        agent_name:str,
        user_request:str,
        tool_context : ToolContext
):
    if agent_name not in AGENT_MAP:
        available = ", ".join(AGENT_MAP.keys())
    agent_url = AGENT_MAP[agent_name]
    run_config = tool_context.run_config

    custom_metadata = run_config.custom_metadata
    session_state_key = f"{agent_name}_session_id"
    category = custom_metadata.get('category','')
    part_nm = custom_metadata.get('part_nm','')
    existing_session_id = tool_context.state.get(session_state_key)
    designer = custom_metadata.get('designer','')
    source_gcs_path = custom_metadata.get('source_gcs_path','')
    chg_req_gcs_path = custom_metadata.get('chg_req_gcs_path','')
    chat_id = custom_metadata.get('chat_id','')
    date = custom_metadata.get('date','')
    
    if existing_session_id:
        metadata['version_session_id'] = existing_session_id
    request_payload = {
        "message":{
            "messageId":str(uuid.uuid4()),
            "role":"ROLE_USER",
            "content":[{"text":user_request}]
        },
        "metadata":{
            "category": category,
            "part_nm": part_nm,
            "designer": designer,
            "source_gcs_path": source_gcs_path,
            "chg_req_gcs_path": chg_req_gcs_path,
            "chat_id": chat_id,
            "date": date,
        },
        'configuration' : {"blocking" : True}
    }
    try:
        auth_token = get_goole_auth_token()
        async with httpx.AsyncClient(timeout=600) as client:
            response = await client.post(
                f"{agent_url}/message:send",
                json = request_payload,
                headers = {
                    "Authorization": f"Bearer {auth_token}",
                    "Content-Type": "application/json"
                })
            reponse.raise_for_status()
            result = response.json()
            new_session_id = _extract_session_id(result)
            if new_session_id:
                tool_context.state[session_state_key] = new_session_id
            return _parse_response(result,agent_name)
    except httpx.HTTPError as e:
        raise UnsupportedOperationError(f"Error communicating with agent {agent_name}: {str(e)}")
    
def _extract_session_id(response: dict) -> str | None:
    if "task" not in response:
        return None
    task = response['task']
    for artifact in task['artifacts']:
        if artifact.get("name") == "vertex_session_id":
            parts = artifact.get("parts",[])
            if parts and "text" in parts[0]:
                return parts[0]["text"].strip()
    return None
def _parse_response(response: dict, agent_name: str) -> str:
    task = response['task']
    results = []
    for artifact in task['artifacts']:
        artifact_name = artifact.get("name",'result')
        if artifact_name =='vertex_session_id':
            continue
        parts = artifact.get("parts",[])
        for i, part in enumerate(parts):
            if "text" in part:
                text = part['text']
                try:
                    import json
                    parsed = json.loads(text)
                    formatted = json.dumps(parsed,indent=2,ensure_ascii=False)
                    results.append(f"[{artifact_name} - part {i+1}]\n{formatted}")
                except json.JSONDecodeError:
                    results.append(f"[{artifact_name} - part {i+1}]\n{text}")
    if results:
        return f"Response from {agent_name}:\n" + "\n\n".join(results)
    return f"No content"
delegate_tool = FunctionTool(delegate_agent)

def create_supervisor_agent() -> LlmAgent :

    SupervisorAgent = LlmAgent(
        name = "SupervisorAgent",
        model = "gemmini-2.5-flash",
        description = "슈퍼바이저",
        instruction = SUPERVISOR_AGENT_INSTRUCTION,
        output_schema = SupervisorOutputSchema,
        tools = [delegate_tool]
    )
    return SupervisorAgent

from vertexai import agent_engines
from functools import partial
app = agent_engines.AdkApp(
    agent = agent,
    enable_tracing=True,
    session_service_builder=partial(
        VertexAiSessionService,
        project= PROJECT_ID,
        location = LOCATION,
        agent_engine_id = AGENT_ENGINE_ID
    ),
    artifact_service_builder=partial(
        GcsArtifactService,
        bucket_name = BUCKET_NAME
    ),
    memory_service_builder=partial(
        VertexAiMemoryBankService,
        project = PROJECT_ID,
        location= LOCATION,
        agent_engine_id = AGENT_ENGINE_ID
)
)

How to do postman or http json SSE communication in that code, what additional code do you need before distributing it

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: To receive a streaming response, you need to set blocking: false in your request payload and update your client-side code to handle a stream of Server-Sent Events (SSE) instead of a single JSON response.

---

Hello! I can help with your question about enabling streaming responses from your AdkApp.

Your current delegate_agent function is making a blocking request, which is why you receive the full response at the end. To switch to a streaming model, you need to make changes in two areas: the request you send and how you handle the response.

1. Modify the Request Payload

In your request_payload, you are setting blocking to True. To initiate a non-blocking, streaming response from the Agent Engine, you should set it to false.

request_payload = {
    "message": {
        # ... your message properties
    },
    "metadata": {
        # ... your metadata properties
    },
    "configuration" : {"blocking" : False}  # Change this to false
}

2. Update Client to Handle SSE Stream

Your current code uses await client.post(...) followed by response.json(), which waits for the entire response body. To handle a stream, you should use client.stream() and asynchronously iterate over the response lines. Each event in an SSE stream is typically prefixed with data:.

Here is a revised version of your delegate_agent function that handles streaming:

import json
import httpx

# ... (keep your AGENT_MAP, get_google_auth_token, etc.)

async def delegate_agent(
        agent_name:str,
        user_request:str,
        tool_context : ToolContext
):
    if agent_name not in AGENT_MAP:
        raise ValueError(f"Unknown agent: {agent_name}")

    agent_url = AGENT_MAP[agent_name]
    run_config = tool_context.run_config
    custom_metadata = run_config.custom_metadata
    session_state_key = f"{agent_name}_session_id"

    # ... (your metadata extraction logic)

    request_payload = {
        "message":{
            "messageId": str(uuid.uuid4()),
            "role": "ROLE_USER",
            "content": [{"text": user_request}]
        },
        "metadata":{
            # ... (your metadata properties)
        },
        "configuration" : {"blocking" : False} # Enable streaming
    }

    try:
        auth_token = get_google_auth_token()
        full_response_text = []

        async with httpx.AsyncClient(timeout=600) as client:
            async with client.stream(
                "POST",
                f"{agent_url}/message:send",
                json=request_payload,
                headers={
                    "Authorization": f"Bearer {auth_token}",
                    "Content-Type": "application/json",
                    "Accept": "text/event-stream" # Indicate we accept SSE
                }
            ) as response:
                response.raise_for_status()
                print(f"Connection opened with {agent_name}")

                async for line in response.aiter_lines():
                    if line.startswith("data:"):
                        data_str = line[len("data:"):].strip()
                        if not data_str:
                            continue
                        try:
                            event = json.loads(data_str)

                            # Example of extracting session_id from a stream artifact
                            new_session_id = _extract_session_id_from_event(event)
                            if new_session_id:
                                tool_context.state[session_state_key] = new_session_id

                            # Example of parsing text from the stream
                            text_chunk = _parse_text_from_event(event)
                            if text_chunk:
                                # You can yield or process each chunk here
                                print(text_chunk, end="", flush=True)
                                full_response_text.append(text_chunk)

                        except json.JSONDecodeError:
                            print(f"\n[Warning: Could not decode JSON from stream: {data_str}]")

        return f"Final response from {agent_name}:\n{' '.join(full_response_text)}"

    except httpx.HTTPError as e:
        raise UnsupportedOperationError(f"Error communicating with agent {agent_name}: {str(e)}")

def _extract_session_id_from_event(event: dict) -> str | None:
    # This is an example; the actual structure may vary.
    # You need to inspect the event structure to find the session ID.
    artifacts = event.get('task', {}).get('artifacts', [])
    for artifact in artifacts:
        if artifact.get("name") == "vertex_session_id":
            parts = artifact.get("parts", [])
            if parts and "text" in parts[0]:
                return parts[0]["text"].strip()
    return None

def _parse_text_from_event(event: dict) -> str | None:
    # This is an example for parsing text from a streaming event.
    # The actual structure depends on the agent's output.
    # You'll need to inspect the events to build the correct parsing logic.
    artifacts = event.get('task', {}).get('artifacts', [])
    for artifact in artifacts:
        # Skip session ID artifacts
        if artifact.get("name") == 'vertex_session_id':
            continue
        parts = artifact.get("parts", [])
        for part in parts:
            if "text" in part:
                return part['text']
    return None

Summary of Changes:

1. Blocking Configuration: Changed {"blocking": True} to {"blocking": False}. [16]
2. HTTP Client: Switched from client.post to client.stream to establish a persistent connection. [8]
3. Response Iteration: Used response.aiter_lines() to read the incoming SSE stream line by line. [8]
4. Event Parsing: Each line starting with data: is treated as a JSON event that needs to be parsed. Your previous _parse_response and _extract_session_id functions need to be adapted to work with the event-based structure of a stream, as shown in the examples _parse_text_from_event and _extract_session_id_from_event. The exact JSON structure of each event may vary, so you should inspect the raw data to build robust parsing logic.

For more details on how streaming is handled within ADK, you can refer to the documentation on RunConfig and streaming modes. [3]

[3] https://google.github.io/adk-docs/runtime/runconfig.html
[8] https://google.github.io/adk-docs/adk-python/src/google/adk/cli/conformance/adk_web_server_client.py.html
[16] https://google.github.io/adk-docs/runtime/api-server.html

[ktk94]

'Blocking' is just a request to the tool, and all I want is that the Supervisor Agent's response will be streamed, including inference.