langgraph_agent_toolkit.service.routes module

async langgraph_agent_toolkit.service.routes.info(request)[source][source]
Parameters:

request (Request)

Return type:

ServiceMetadata

async langgraph_agent_toolkit.service.routes.invoke(user_input, agent_id=None, request=None)[source][source]

Invoke an agent and return its final response.

Use the default agent when agent_id is not provided. Use thread_id to persist a multi-turn conversation. Messages include the run_id keyword argument for feedback recording.

Parameters:
  • user_input (UserInput)

  • agent_id (str)

  • request (Request)

Return type:

ChatMessage

async langgraph_agent_toolkit.service.routes.stream(user_input, agent_id=None, request=None)[source][source]

Stream agent responses, including messages and tokens.

Use the default agent when agent_id is not provided. Use thread_id to persist a multi-turn conversation. Messages include the run_id keyword argument for feedback recording. Set stream_tokens=false to exclude token output.

Parameters:
  • user_input (StreamInput)

  • agent_id (str | None)

  • request (Request)

Return type:

ClosingStreamingResponse

async langgraph_agent_toolkit.service.routes.stream_jsonl(user_input, agent_id=None, request=None)[source][source]

Stream agent responses as JSON Lines with one StreamChunk per line.

This typed, OpenAPI-documented endpoint is an alternative to SSE /stream. It emits token chunks when stream_tokens is true and full message chunks. A failure emits a final error chunk. The response body end terminates the stream.

Parameters:
  • user_input (StreamInput)

  • agent_id (str | None)

  • request (Request)

Return type:

ClosingStreamingResponse

async langgraph_agent_toolkit.service.routes.feedback(feedback, agent_id=None, request=None)[source][source]

Record feedback for a run on the configured observability platform.

The agent configuration selects the observability platform.

Parameters:
  • feedback (Feedback)

  • agent_id (str | None)

  • request (Request)

Return type:

FeedbackResponse

async langgraph_agent_toolkit.service.routes.history(input=Depends(), agent_id=None, request=None)[source][source]

Get short-term chat history for one thread.

Parameters:
Return type:

ChatHistory

async langgraph_agent_toolkit.service.routes.clear_history(input, agent_id=None, request=None)[source][source]

Delete one thread’s checkpoints. Keep long-term memory.

Parameters:
Return type:

ClearHistoryResponse

async langgraph_agent_toolkit.service.routes.add_messages(input, agent_id=None, request=None)[source][source]

Add messages to one thread’s short-term history.

Parameters:
Return type:

AddMessagesResponse

async langgraph_agent_toolkit.service.routes.redirect_to_docs()[source][source]
Return type:

RedirectResponse

async langgraph_agent_toolkit.service.routes.health_check()[source][source]

Health check endpoint.

Return type:

HealthCheck

async langgraph_agent_toolkit.service.routes.liveness_probe(request)[source][source]

Return the Kubernetes liveness probe response.

This probe confirms that the process responds. Kubernetes restarts the process when the probe fails.

Parameters:

request (Request)

async langgraph_agent_toolkit.service.routes.readiness_probe(request)[source][source]

Return the Kubernetes readiness probe response.

This probe checks that the service initializes agents successfully. Kubernetes routes traffic to the pod only after this returns 200.

Parameters:

request (Request)

async langgraph_agent_toolkit.service.routes.startup_probe(request)[source][source]

Return the Kubernetes startup probe response.

This probe checks whether the application finished initialization. It supports slow-starting containers. Kubernetes checks it before liveness and readiness probes.

Parameters:

request (Request)

async langgraph_agent_toolkit.service.routes.db_health_check(request)[source][source]

Database pool health check endpoint.

Parameters:

request (Request)

Return type:

DatabaseHealthResponse