from typing import Any, Dict, List, Literal, NotRequired
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
from langgraph_agent_toolkit.core.settings import settings
from langgraph_agent_toolkit.helper.constants import (
DEFAULT_MODEL_PARAMETER_VALUES,
get_default_agent,
)
[docs]
class AgentInfo(BaseModel):
"""Info about an available agent."""
key: str = Field(
description="Agent key.",
examples=["langgraph-supervisor-agent"],
)
description: str = Field(
description="Description of the agent.",
examples=["A research assistant."],
)
[docs]
class ChatMessage(BaseModel):
"""Message in a chat."""
type: Literal["human", "ai", "tool", "custom"] = Field(
description="Role of the message.",
examples=["human", "ai", "tool", "custom"],
)
content: str | Dict[str, Any] | List[str | Dict[str, Any]] = Field(
description="Content of the message.",
examples=["Hello, world!"],
)
tool_calls: list[ToolCall] = Field(
description="Tool calls in the message.",
default=[],
)
tool_call_id: str | None = Field(
description="Tool call that this message is responding to.",
default=None,
examples=["call_Jja7J89XsjrOLA5r!MEOW!SL"],
)
run_id: str | None = Field(
description="Run ID of the message.",
default=None,
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
response_metadata: dict[str, Any] = Field(
description="Response metadata. For example: response headers, logprobs, token counts.",
default={},
)
custom_data: dict[str, Any] = Field(
description="Custom message data.",
default={},
)
[docs]
def pretty_repr(self) -> str:
"""Get a pretty representation of the message."""
base_title = self.type.title() + " Message"
padded = " " + base_title + " "
sep_len = (80 - len(padded)) // 2
sep = "=" * sep_len
second_sep = sep + "=" if len(padded) % 2 else sep
title = f"{sep}{padded}{second_sep}"
return f"{title}\n\n{self.content}"
[docs]
def pretty_print(self) -> None:
print(self.pretty_repr()) # noqa: T201
[docs]
class Feedback(BaseModel):
"""Feedback for a run, to record to LangSmith."""
run_id: str = Field(
description="Run ID to record feedback for.",
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
key: str = Field(
description="Feedback key.",
examples=["human-feedback-stars"],
)
score: float = Field(
description="Feedback score.",
examples=[0.8],
)
user_id: str | None = Field(
description="User ID to associate with the feedback.",
default=None,
examples=["521c0a60-ea75-43fa-a793-a4cf11e013ae"],
)
kwargs: dict[str, Any] = Field(
description="Additional feedback kwargs, passed to LangSmith.",
default={},
examples=[{"comment": "In-line human feedback"}],
)
[docs]
class FeedbackResponse(BaseModel):
"""Response after recording feedback."""
status: Literal["success"] = "success"
run_id: str = Field(
description="Run ID for which feedback was recorded.",
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
message: str = Field(
description="Descriptive message about the feedback operation.",
default="Feedback recorded successfully.",
)
[docs]
class AddMessagesResponse(BaseModel):
"""Response after adding messages to the chat history."""
status: Literal["success"] = "success"
thread_id: str | None = Field(
description="Thread ID for which the message was added.",
default=None,
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
user_id: str | None = Field(
description="User ID associated with the message.",
default=None,
examples=["521c0a60-ea75-43fa-a793-a4cf11e013ae"],
)
message: str = Field(
description="Descriptive message about the operation.",
default="Messages added successfully.",
)
[docs]
class ClearHistoryInput(BaseModel):
"""Input for clearing messages from the chat history."""
thread_id: str | None = Field(
description="Thread ID to persist and continue a multi-turn conversation.",
default=None,
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
user_id: str | None = Field(
description="User ID to persist in observability platform and share long-term memory.",
default=None,
examples=["521c0a60-ea75-43fa-a793-a4cf11e013ae"],
)
[docs]
class ClearHistoryResponse(BaseModel):
"""Response after clearing messages from the chat history."""
status: Literal["success"] = "success"
thread_id: str | None = Field(
description="Thread ID for which the messages were cleared.",
default=None,
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
user_id: str | None = Field(
description="User ID associated with the operation.",
default=None,
examples=["521c0a60-ea75-43fa-a793-a4cf11e013ae"],
)
message: str = Field(
description="Descriptive message about the operation.",
default="Messages cleared successfully.",
)
[docs]
class ChatHistoryInput(BaseModel):
"""Input for retrieving chat history."""
thread_id: str | None = Field(
description="Thread ID to persist and continue a multi-turn conversation.",
default=None,
examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
)
user_id: str | None = Field(
description="User ID to persist in observability platform and share long-term memory.",
default=None,
examples=["521c0a60-ea75-43fa-a793-a4cf11e013ae"],
)
[docs]
class ChatHistory(BaseModel):
messages: list[ChatMessage]
[docs]
class HealthCheck(BaseModel):
"""Response model to validate and return when performing a health check."""
content: str = Field(
...,
description="Health status of the service.",
examples=["healthy"],
)
version: str = Field(
...,
description="Version of the service.",
examples=["1.0.0"],
)
[docs]
class LivenessResponse(BaseModel):
"""Response model for liveness probe - checks if the process is alive."""
status: Literal["alive", "unhealthy"] = Field(
description="Liveness status of the service.",
examples=["alive"],
)
version: str = Field(
description="Version of the service.",
examples=["1.0.0"],
)
[docs]
class ReadinessResponse(BaseModel):
"""Response model for readiness probe - checks if service can accept traffic."""
status: Literal["ready", "not_ready"] = Field(
description="Readiness status of the service.",
examples=["ready"],
)
version: str = Field(
description="Version of the service.",
examples=["1.0.0"],
)
initialized_agents: List[str] = Field(
default=[],
description="List of successfully initialized agent IDs.",
examples=[["react_agent", "chatbot_agent"]],
)
message: str = Field(
default="",
description="Additional information about readiness status.",
examples=["All agents initialized successfully"],
)
[docs]
class StartupResponse(BaseModel):
"""Response model for startup probe - checks if application has started."""
status: Literal["started", "starting"] = Field(
description="Startup status of the service.",
examples=["started"],
)
version: str = Field(
description="Version of the service.",
examples=["1.0.0"],
)
message: str = Field(
default="",
description="Additional information about startup status.",
examples=["Application startup complete"],
)
[docs]
class DatabaseHealthResponse(BaseModel):
"""Response model for database health check."""
status: Literal["healthy", "exhausted", "no_pool", "error"] = Field(
description="Database connection pool status.",
examples=["healthy"],
)
message: str | None = Field(
default=None,
description="Additional information about the database status.",
)
pool_size: int | None = Field(
default=None,
description="Total size of the connection pool.",
)
pool_available: int | None = Field(
default=None,
description="Number of available connections in the pool.",
)
requests_waiting: int | None = Field(
default=None,
description="Number of requests waiting for a connection.",
)
connections_num: int | None = Field(
default=None,
description="Current number of connections.",
)