- langgraph_agent_toolkit.service.utils.verify_bearer(http_auth)[source][source]
- Parameters:
http_auth (Annotated[HTTPAuthorizationCredentials | None, Depends(dependency=<fastapi.security.http.HTTPBearer object at 0x7fd9d4772210>, use_cache=True, scope=None)])
- Return type:
None
- langgraph_agent_toolkit.service.utils.get_agent_executor(request)[source][source]
Get the AgentExecutor instance that was initialized in lifespan.
- Parameters:
request (Request)
- Return type:
- langgraph_agent_toolkit.service.utils.get_agent(request, agent_id)[source][source]
Get an agent by its ID from the initialized AgentExecutor.
- langgraph_agent_toolkit.service.utils.get_all_agent_info(request)[source][source]
Get information about all available agents from the initialized AgentExecutor.
- Parameters:
request (Request)
- async langgraph_agent_toolkit.service.utils.message_generator(stream_input, request, agent_id)[source][source]
Generate messages from an agent.
- Parameters:
stream_input (StreamInput)
request (Request)
agent_id (str)
- Return type:
AsyncGenerator[str, None]
- async langgraph_agent_toolkit.service.utils.jsonl_message_generator(stream_input, request, agent_id)[source][source]
Yield typed StreamChunk objects for a JSON Lines (NDJSON) agent stream.
The JSON Lines twin of message_generator: instead of formatting SSE frames, it emits typed StreamChunk objects that FastAPI serializes as one JSON object per line. There is no [DONE] sentinel — for NDJSON the end of the response body terminates the stream. Mid-stream failures are surfaced as a final type=”error” chunk (never an unhandled raise, which would corrupt an already-started response).
- Parameters:
stream_input (StreamInput)
request (Request)
agent_id (str)
- Return type:
AsyncGenerator[StreamChunk, None]