langgraph_agent_toolkit.service.utils module
- langgraph_agent_toolkit.service.utils.verify_bearer(http_auth, request)[source][source]
- Parameters:
http_auth (Annotated[HTTPAuthorizationCredentials | None, Depends(dependency=<fastapi.security.http.HTTPBearer object at 0x7f147a35d0d0>, use_cache=True, scope=None)])
request (Request)
- Return type:
None
- langgraph_agent_toolkit.service.utils.get_agent_executor(request)[source][source]
Get the AgentExecutor initialized during lifespan.
- Parameters:
request (Request)
- Return type:
- langgraph_agent_toolkit.service.utils.get_agent(request, agent_id)[source][source]
Get an agent by ID from the initialized AgentExecutor.
- Parameters:
request (Request)
agent_id (str)
- Return type:
- langgraph_agent_toolkit.service.utils.get_all_agent_info(request)[source][source]
Get all agent information from the initialized AgentExecutor.
- Parameters:
request (Request)
- async langgraph_agent_toolkit.service.utils.jsonl_message_generator(stream_input, request, agent_id, public_thread_id=None)[source][source]
Yield typed events and close execution when the consumer disconnects.
- Parameters:
stream_input (StreamInput)
request (Request)
agent_id (str)
public_thread_id (str | None)
- Return type:
AsyncGenerator[StreamChunk, None]
- async langgraph_agent_toolkit.service.utils.message_generator(stream_input, request, agent_id, public_thread_id=None)[source][source]
Encode typed events as SSE frames.
- Parameters:
stream_input (StreamInput)
request (Request)
agent_id (str)
public_thread_id (str | None)
- Return type:
AsyncGenerator[str, None]