How to use vertexai.Client to connect to local adk api_server ?

[markns]

I'm using the vertexai.Client to connect my UI code to my adk agent running on vertex agent engine.

self.client = vertexai.Client(location=settings.REASONING_ENGINE_LOCATION)
self.agent_engine = self.client.agent_engines.get(name=settings.REASONING_ENGINE_ID)

async for event in self.agent_engine.async_stream_query(
        user_id=user_id, session_id=session_id, message=message
):
    ...

However, for development iteration speed, I would like to connect the same UI code to an agent running locally (presumably with adk api_server).

How should I adjust the client code to connect to the api_server?

[adk-bot]

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

TLDR: The vertexai.Client is designed to connect to managed agents on Google Cloud's Vertex AI platform and cannot be used to connect to a local adk api_server. For local development, you should make requests directly to the REST API exposed by the local server using a standard HTTP client.

---

Hello! That's a great question about streamlining local development with ADK.

The vertexai.Client you are using is part of the Vertex AI SDK and is specifically designed to interact with agents deployed on the managed Vertex AI platform on Google Cloud. [12, 18] It handles authentication and communication with the cloud-based Agent Engine service.

For local development, the adk api_server command starts a local FastAPI web server that exposes your agent through a RESTful API, typically running at http://localhost:8000. [2]

To connect your UI to this local server, you should use a standard HTTP client to send requests to the appropriate endpoint. The equivalent of the agent_engine.async_stream_query method is the /run_sse endpoint, which provides a stream of Server-Sent Events (SSE).

Here is an example of how you could adapt your code to connect to the local server using the httpx library. You would need to specify the app_name, which corresponds to your agent's directory name.

import httpx
import json

# Configuration for your local agent
LOCAL_AGENT_URL = "http://localhost:8000/run_sse"
AGENT_APP_NAME = "your_agent_directory_name" # Change this to your agent's app name

async def stream_local_agent(user_id: str, session_id: str, message: str):
    """
    Connects to a local adk api_server and streams the response.
    """
    payload = {
        "app_name": AGENT_APP_NAME,
        "user_id": user_id,
        "session_id": session_id,
        "message": message,
    }
    headers = {
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
    }

    try:
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream("POST", LOCAL_AGENT_URL, json=payload, headers=headers) as response:
                # Ensure the request was successful
                response.raise_for_status()
                print("Successfully connected to the local agent stream.")
                async for line in response.aiter_lines():
                    if line.startswith("data:"):
                        try:
                            # The event data is in a JSON string, remove "data:" prefix
                            event_data = json.loads(line[5:])
                            # Now you can process the event like you did before
                            print(event_data)
                            # ... your event processing logic here ...
                        except json.JSONDecodeError:
                            print(f"Could not decode JSON from line: {line}")

    except httpx.RequestError as e:
        print(f"An error occurred while requesting {e.request.url!r}.")
        print("Please ensure your 'adk api_server' is running.")


# Example usage:
# import asyncio
# asyncio.run(stream_local_agent(user_id="test_user", session_id="test_session", message="Hello agent!"))

By using a conditional check in your code (e.g., based on an environment variable), you can easily switch between the vertexai.Client for your deployed agent and this local streaming method for development.

I hope this helps you accelerate your development workflow!

---

References:

• [1] Use the API Server. https://storage.googleapis.com/adk-knowledge-bucket-cloud-vi-test/adk-docs/docs/runtime/api-server.html
• [2] Express Mode. https://storage.googleapis.com/adk-knowledge-bucket-cloud-vi-test/adk-docs/docs/sessions/express-mode.html
• [3] Bidirectional Streaming Development Guide - Part 1: Platforms. https://storage.googleapis.com/adk-knowledge-bucket-cloud-vi-test/adk-docs/docs/streaming/dev-guide/part1.html

[markns]

Hmm, that's really bad.
Is there no client for the adk api_server?