Generative AI Google Brain

gemini-agents-genai-experience-concierge-langgraph-demo-backend-notebooks-langgraph-remote-agent

3 min read

Query a Remote LangGraph Agent Server

Source notebook

Repo path: gemini/agents/genai-experience-concierge/langgraph-demo/backend/notebooks/langgraph-remote-agent.ipynb · Open on GitHub · intermediate

Queries a local or Cloud Run LangGraph RemoteGraph agent and displays streamed agent outputs and state.

Summary

This notebook demonstrates how to connect to a deployed LangGraph agent server with the standard RemoteGraph client. It configures a local or Cloud Run agent endpoint, streams node updates and custom text chunks, formats heterogeneous agent events, and inspects session state and history.

Key code patterns

Create RemoteGraph client

agent_name = "task-planner"
agent_url = f"http://127.0.0.1:3000/{agent_name}"
id_token = None
 
graph = remote.RemoteGraph(
    agent_name,
    url=agent_url,
    headers={"Authorization": f"Bearer {id_token}"} if id_token else {},
)

Configures a remote LangGraph client for either local development or an authenticated deployed endpoint.

Stream node updates

for chunk in graph.stream(
    input={"current_turn": {"user_input": "hi"}},
    config={"configurable": {"thread_id": test_thread}},
    stream_mode="updates",
):
    print(chunk)

Shows how to stream graph node updates for a specific conversation thread.

Stream custom chunks

for stream_mode, chunk in graph.stream(
    input={"current_turn": {"user_input": "what products does Cymbal Retail sell?"}},
    config={"configurable": {"thread_id": test_thread}},
    stream_mode=["updates", "custom"],
):
    if stream_mode == "custom":
        current_source, text = handle_chunk(chunk, task_idx)

Uses LangGraph custom stream mode to consume text, classifications, function calls, plans, task results, and errors.

Inspect graph state

snapshot = graph.get_state(
    config={"configurable": {"thread_id": test_thread}}
)
 
snapshot_list = list(
    graph.get_state_history(config={"configurable": {"thread_id": test_thread}})
)

Retrieves current and historical state snapshots for the same thread id.

Models & APIs used

  • APIs / services: Cloud Run
  • SDKs / libraries: langgraph

When to use this

Use this pattern when building or testing a frontend or client that needs to call a hosted LangGraph agent and handle streamed events.

Gotchas & caveats

  • The notebook defaults to a local server at http://127.0.0.1:3000/{agent_name}.
  • A deployed Cloud Run endpoint may require an identity token from gcloud auth print-identity-token.
  • The custom chunk handler is described as useful for demo purposes but not very practical in practice.
  • The stream handling assumes custom chunks are dictionaries.

Best practices

  • Use a unique thread_id to keep each test session isolated.
  • Pass an Authorization bearer header only when an id_token is available.
  • Handle multiple chunk shapes explicitly, including text, responses, guardrail classifications, router classifications, function calls, function responses, plans, executed tasks, errors, and unhandled keys.
  • Use get_state and get_state_history to inspect the remote agent session after streaming.

Graph View

Backlinks