google / google/adk-python

bidi_stream_query documented but not implemented in AdkApp.register_operations()

Abierto
#5,611 2 comentarios 0 reacciones 1 asignado Reclamado por @sanketpatil06 Ver en GitHub
live needs review
Lenguaje dominante
Python
Estrellas
21.5k
Forks
4k
Merge medio
1 d 14 h
PR fusionados (30 d)
37

Descripción

## Summary

The Google Cloud documentation references `bidi_stream_query` as a supported streaming method for deployed ADK agents on Vertex AI Agent Runtime, but the method does not exist in any public version of the ADK SDK.

## Expected behavior

`bidi_stream_query` should be available as a WebSocket-based bidirectional streaming method on a deployed `AdkApp`, as described in the official discussion post:
https://discuss.google.dev/t/deploy-bidirectional-streaming-agents-with-vertex-ai-agent-engine-and-live-api/266087

## Actual behavior

`AdkApp.register_operations()` returns the following — no `bidi_stream_query`:

```python
{
"": ["get_session", "list_sessions", "create_session", "delete_session"],
"async": ["async_get_session", "async_list_sessions", "async_create_session", "async_delete_session", "async_add_session_to_memory", "async_search_memory"],
"stream": ["stream_query"],
"async_stream": ["async_stream_query", "streaming_agent_run_with_events"],
}
```

Attempting to invoke `bidi_stream_query` on a deployed engine via `client.aio.live.agent_engines.connect()` returns:
```
bidi_stream_query not found. Available methods are: []
```

## Versions tested

- `google-adk==1.32.0` — `bidi_stream_query` absent from `register_operations()`
- `google-adk==2.0.0b1` (beta) — same result, identical `register_operations()` output

Verified locally by importing `AdkApp` and calling `register_operations()` directly:

```python
from vertexai.agent_engines.templates.adk import AdkApp
print(AdkApp.register_operations(None))
# Returns the dict above — no bidi_stream_query
```

## Workaround

Ended up using `streaming_agent_run_with_events` (registered under `async_stream`), which works but:
- Takes a single `request_json: str` parameter (a JSON-serialised `_StreamRunRequest`) rather than a `LiveRequest` object
- Yields `_StreamingRunResponse` dicts rather than ADK `Event` objects directly
- Each yielded chunk is nested as `chunk["output"]["events"]`

Example working call:
```python
request_json = json.dumps({
"message": {"role": "user", "parts": [{"text": message}]},
"user_id": customer_id,
"session_id": session_id,
})
async for chunk in engine.streaming_agent_run_with_events(request_json=request_json):
events = chunk.get("output", chunk).get("events", [])
for event in events:
if event.get("author") == "my_agent":
for part in event.get("content", {}).get("parts", []):
print(part.get("text", ""))
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.