Source code for langgraph_agent_toolkit.schema.schema
from typing import Any, Dict, List, Literal, NotRequired
from pydantic import BaseModel, Field, field_validator
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 ServiceMetadata(BaseModel):
"""Metadata about the service including available agents and models."""
agents: list[AgentInfo] = Field(
description="List of available agents.",
)
default_agent: str = Field(
description="Default agent used when none is specified.",
examples=[get_default_agent()],
)
[docs]
class UserComplexInput(BaseModel):
"""Basic user input for the agent, supporting dynamic fields."""
message: str | list[dict[str, Any]] | None = Field(
default=None,
description=(
"User input to the agent: either plain text, or a list of LangChain content blocks for "
"multimodal input. Each block is {'type': 'text'|'image'|'file'|'audio'|'video', ...} with "
"'text', a 'url', or 'base64'+'mime_type'. LangChain translates blocks to the provider's "
"native format, so the chosen model must support the modality."
),
examples=[
"What is the weather in Tokyo?",
[
{"type": "text", "text": "Describe this image."},
{"type": "image", "url": "https://example.com/image.jpg"},
],
[
{"type": "text", "text": "Summarize this document."},
{"type": "file", "base64": "<base64-bytes>", "mime_type": "application/pdf"},
],
],
)
model_config = {
"extra": "allow" # allow unknown fields
}
@field_validator("message")
@classmethod
def _validate_content_blocks(cls, value: "str | list[dict[str, Any]] | None"):
"""Lightly validate multimodal content blocks; LangChain does the deep validation downstream."""
if not isinstance(value, list):
return value
allowed = {"text", "image", "file", "audio", "video"}
# Any recognized content source. Kept permissive on purpose so valid alternative forms
# (file_id, id, source_type, ...) are not rejected — only blocks that carry no content
# reference at all (e.g. {"type": "image"}) fail here; LangChain does the deep validation.
content_keys = {"url", "base64", "data", "file_id", "id", "source_type", "source", "path"}
media_count = 0
for i, block in enumerate(value):
if not isinstance(block, dict) or "type" not in block:
raise ValueError(f"content block {i} must be a dict with a 'type' field")
btype = block["type"]
if btype not in allowed:
raise ValueError(f"content block {i} has unsupported type {btype!r}; expected one of {sorted(allowed)}")
if btype == "text":
if not isinstance(block.get("text"), str):
raise ValueError(f"content block {i} of type 'text' must include a string 'text' field")
else:
media_count += 1
if block.get("base64") and not isinstance(block.get("mime_type"), str):
raise ValueError(
f"content block {i} of type {btype!r} must include a string 'mime_type' when using 'base64'"
)
if not any(block.get(k) for k in content_keys):
raise ValueError(
f"content block {i} of type {btype!r} must include a content source "
"(e.g. a 'url', or 'base64' + 'mime_type')"
)
max_attachments = settings.MULTIMODAL_MAX_ATTACHMENTS
if max_attachments is not None and media_count > max_attachments:
raise ValueError(
f"too many attachments: {media_count} (max {max_attachments}); "
f"adjust MULTIMODAL_MAX_ATTACHMENTS to change the limit"
)
return value
[docs]
class UserInput(BaseModel):
"""Basic user input for the agent."""
input: UserComplexInput = Field(
description="Structured input from the user, including a message and optional dynamic fields.",
examples=[
{
"message": "What is the weather in Tokyo?",
}
],
)
model_name: str | None = Field(
title="Model",
description="LLM Model Name to use for the agent.",
default=None,
examples=["gpt-3.5-turbo", "gpt-4o"],
)
model_provider: str | None = Field(
title="Model Provider",
description="LLM Model Provider to use for the agent.",
default=None,
examples=["openai", "anthropic"],
)
model_config_key: str | None = Field(
title="Model Configuration Key",
description="Key for predefined model configuration in MODEL_CONFIGS.",
default=None,
examples=["gpt4o", "gemini"],
)
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"],
)
agent_config: dict[str, Any] = Field(
description="Additional configuration to pass through to the agent",
default={},
examples=[
{
"checkpointer_params": {"k": 6},
**DEFAULT_MODEL_PARAMETER_VALUES,
},
],
)
recursion_limit: int | None = Field(
description="Recursion limit for the agent.",
default=None,
examples=[settings.DEFAULT_RECURSION_LIMIT],
)
[docs]
class StreamInput(UserInput):
"""User input for streaming the agent's response."""
stream_tokens: bool = Field(
description="Whether to stream LLM tokens to the client.",
default=True,
)
[docs]
class ToolCall(TypedDict):
"""Represents a request to call a tool."""
name: str
"""The name of the tool to be called."""
args: dict[str, Any]
"""The arguments to the tool call."""
id: str | None
"""An identifier associated with the tool call."""
type: NotRequired[Literal["tool_call"]]
[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]
class StreamChunk(BaseModel):
"""A single chunk of a JSON Lines (NDJSON) agent stream.
One StreamChunk is emitted per line by the ``/stream/jsonl`` endpoint:
- ``type="token"`` -> ``content`` is an incremental token string.
- ``type="message"`` -> ``content`` is a complete :class:`ChatMessage`.
- ``type="error"`` -> ``content`` is an error description string.
"""
type: Literal["token", "message", "error"] = Field(description="The kind of chunk.")
content: str | ChatMessage = Field(description="Token text, a full ChatMessage, or an error string.")
[docs]
class ErrorResponse(BaseModel):
"""Standard error response body returned by the service's exception handlers."""
detail: str = Field(description="Human-readable error message.")
error_code: str | None = Field(default=None, description="Stable machine-readable error code, when present.")
[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 MessageInput(BaseModel):
"""Input for a message to be added to the chat history."""
type: Literal["human", "ai", "tool", "custom"] = Field(
description="Role of the message.",
examples=["human", "ai", "tool", "custom"],
)
content: str = Field(
description="Content of the message.",
examples=["Hello, world!"],
)
[docs]
class AddMessagesInput(BaseModel):
"""Input for adding messages to 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"],
)
messages: list[MessageInput] = Field(
description="List of messages to add to the chat history.",
examples=[
[
{
"type": "human",
"content": "Hello, how are you?",
},
{
"type": "ai",
"content": "I'm doing well, thank you! How can I assist you today?",
},
]
],
)
[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 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.",
)