Files
letta-server/letta/schemas/block.py
cthomas ab4ccfca31 feat: add tags support to blocks (#8474)
* feat: add tags support to blocks

* fix: add timestamps and org scoping to blocks_tags

Addresses PR feedback:

1. Migration: Added timestamps (created_at, updated_at), soft delete
   (is_deleted), audit fields (_created_by_id, _last_updated_by_id),
   and organization_id to blocks_tags table for filtering support.
   Follows SQLite baseline pattern (composite PK of block_id+tag, no
   separate id column) to avoid insert failures.

2. ORM: Relationship already correct with lazy="raise" to prevent
   implicit joins and passive_deletes=True for efficient CASCADE deletes.

3. Schema: Changed normalize_tags() from Any to dict for type safety.

4. SQLite: Added blocks_tags to SQLite baseline schema to prevent
   table-not-found errors.

5. Code: Updated all tag row inserts to include organization_id.

🐾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta <noreply@letta.com>

* fix: add ORM columns and update SQLite baseline for blocks_tags

Fixes test failures (CompileError: Unconsumed column names: organization_id):

1. ORM: Added organization_id, timestamps, audit fields to BlocksTags
   ORM model to match database schema from migrations.

2. SQLite baseline: Added full column set to blocks_tags (organization_id,
   timestamps, audit fields) to match PostgreSQL schema.

3. Test: Added 'tags' to expected Block schema fields.

This ensures SQLite and PostgreSQL have matching schemas and the ORM
can consume all columns that the code inserts.

🐾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta <noreply@letta.com>

* revert change to existing alembic migration

* fix: remove passive_deletes and SQLite support for blocks_tags

1. Removed passive_deletes=True from Block.tags relationship to match
   AgentsTags pattern (neither have ondelete CASCADE in DB schema).

2. Removed SQLite branch from _replace_block_pivot_rows_async since
   blocks_tags table is PostgreSQL-only (migration skips SQLite).

🐾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta <noreply@letta.com>

* api sync

---------

Co-authored-by: Letta <noreply@letta.com>
2026-01-19 15:54:38 -08:00

225 lines
8.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from datetime import datetime
from typing import Any, List, Optional
from pydantic import ConfigDict, Field, model_validator
from letta.constants import CORE_MEMORY_BLOCK_CHAR_LIMIT, DEFAULT_HUMAN_BLOCK_DESCRIPTION, DEFAULT_PERSONA_BLOCK_DESCRIPTION
from letta.schemas.enums import PrimitiveType
from letta.schemas.letta_base import LettaBase
# block of the LLM context
class BaseBlock(LettaBase, validate_assignment=True):
"""Base block of the LLM context"""
__id_prefix__ = PrimitiveType.BLOCK.value
# data value
value: str = Field(..., description="Value of the block.")
limit: int = Field(CORE_MEMORY_BLOCK_CHAR_LIMIT, description="Character limit of the block.")
project_id: Optional[str] = Field(None, description="The associated project id.")
# template data (optional)
template_name: Optional[str] = Field(None, description="Name of the block if it is a template.")
is_template: bool = Field(False, description="Whether the block is a template (e.g. saved human/persona options).")
template_id: Optional[str] = Field(None, description="The id of the template.")
base_template_id: Optional[str] = Field(None, description="The base template id of the block.")
deployment_id: Optional[str] = Field(None, description="The id of the deployment.")
entity_id: Optional[str] = Field(None, description="The id of the entity within the template.")
preserve_on_migration: Optional[bool] = Field(False, description="Preserve the block on template migration.")
# context window label
label: Optional[str] = Field(None, description="Label of the block (e.g. 'human', 'persona') in the context window.")
# permissions of the agent
read_only: bool = Field(False, description="Whether the agent has read-only access to the block.")
# metadata
description: Optional[str] = Field(None, description="Description of the block.")
metadata: Optional[dict] = Field({}, description="Metadata of the block.")
hidden: Optional[bool] = Field(
None,
description="If set to True, the block will be hidden.",
)
# def __len__(self):
# return len(self.value)
model_config = ConfigDict(extra="ignore") # Ignores extra fields
@model_validator(mode="before")
@classmethod
def verify_char_limit(cls, data: Any) -> Any:
"""Validate the character limit before model instantiation.
Notes:
- Runs on raw input; do not mutate input.
- For update schemas (e.g., BlockUpdate), `value` and `limit` may be absent.
In that case, only validate when both are provided.
"""
if isinstance(data, dict):
limit = data.get("limit")
value = data.get("value")
# Only enforce the char limit when both are present.
# Pydantic will separately enforce required fields where applicable.
if limit is not None and value is not None and isinstance(value, str):
if len(value) > limit:
error_msg = f"Edit failed: Exceeds {limit} character limit (requested {len(value)})"
raise ValueError(error_msg)
return data
def __setattr__(self, name, value):
"""Run validation if self.value is updated"""
super().__setattr__(name, value)
if name == "value":
# run validation
self.__class__.model_validate(self.model_dump(exclude_unset=True))
class Block(BaseBlock):
"""A Block represents a reserved section of the LLM's context window."""
id: str = BaseBlock.generate_id_field()
# default orm fields
created_by_id: Optional[str] = Field(None, description="The id of the user that made this Block.")
last_updated_by_id: Optional[str] = Field(None, description="The id of the user that last updated this Block.")
# tags - using Optional with default [] to allow None input to become empty list
tags: Optional[List[str]] = Field(default=[], description="The tags associated with the block.")
@model_validator(mode="before")
@classmethod
def normalize_tags(cls, data: dict) -> dict:
"""Convert None tags to empty list."""
if isinstance(data, dict) and data.get("tags") is None:
data["tags"] = []
return data
class BlockResponse(Block):
id: str = Field(
...,
description="The id of the block.",
)
template_name: Optional[str] = Field(
None, description="(Deprecated) The name of the block template (if it is a template).", deprecated=True
)
template_id: Optional[str] = Field(None, description="(Deprecated) The id of the template.", deprecated=True)
base_template_id: Optional[str] = Field(None, description="(Deprecated) The base template id of the block.", deprecated=True)
deployment_id: Optional[str] = Field(None, description="(Deprecated) The id of the deployment.", deprecated=True)
entity_id: Optional[str] = Field(None, description="(Deprecated) The id of the entity within the template.", deprecated=True)
preserve_on_migration: Optional[bool] = Field(
False, description="(Deprecated) Preserve the block on template migration.", deprecated=True
)
read_only: bool = Field(False, description="(Deprecated) Whether the agent has read-only access to the block.", deprecated=True)
hidden: Optional[bool] = Field(None, description="(Deprecated) If set to True, the block will be hidden.", deprecated=True)
class FileBlock(Block):
file_id: str = Field(..., description="Unique identifier of the file.")
source_id: str = Field(..., description="Deprecated: Use `folder_id` field instead. Unique identifier of the source.", deprecated=True)
is_open: bool = Field(..., description="True if the agent currently has the file open.")
last_accessed_at: Optional[datetime] = Field(
None,
description="UTC timestamp of the agents most recent access to this file. Any operations from the open, close, or search tools will update this field.",
)
class Human(Block):
"""Human block of the LLM context"""
label: str = "human"
description: Optional[str] = Field(DEFAULT_HUMAN_BLOCK_DESCRIPTION, description="Description of the block.")
class Persona(Block):
"""Persona block of the LLM context"""
label: str = "persona"
description: Optional[str] = Field(DEFAULT_PERSONA_BLOCK_DESCRIPTION, description="Description of the block.")
DEFAULT_BLOCKS = [Human(value=""), Persona(value="")]
class BlockUpdate(BaseBlock):
"""Update a block"""
limit: Optional[int] = Field(None, description="Character limit of the block.")
value: Optional[str] = Field(None, description="Value of the block.")
project_id: Optional[str] = Field(None, description="The associated project id.")
# tags
tags: Optional[List[str]] = Field(None, description="The tags to associate with the block.")
model_config = ConfigDict(extra="ignore") # Ignores extra fields
class CreateBlock(BaseBlock):
"""Create a block"""
label: str = Field(..., description="Label of the block.")
limit: int = Field(CORE_MEMORY_BLOCK_CHAR_LIMIT, description="Character limit of the block.")
value: str = Field(..., description="Value of the block.")
project_id: Optional[str] = Field(None, description="The associated project id.")
# block templates
is_template: bool = False
template_name: Optional[str] = Field(None, description="Name of the block if it is a template.")
# tags
tags: Optional[List[str]] = Field(None, description="The tags to associate with the block.")
@model_validator(mode="before")
@classmethod
def ensure_value_is_string(cls, data):
"""Convert None value to empty string"""
if data and isinstance(data, dict) and data.get("value") is None:
data["value"] = ""
return data
class CreateHuman(CreateBlock):
"""Create a human block"""
label: str = "human"
class CreatePersona(CreateBlock):
"""Create a persona block"""
label: str = "persona"
class CreateBlockTemplate(CreateBlock):
"""Create a block template"""
is_template: bool = True
class CreateHumanBlockTemplate(CreateHuman):
"""Create a human block template"""
is_template: bool = True
label: str = "human"
class CreatePersonaBlockTemplate(CreatePersona):
"""Create a persona block template"""
is_template: bool = True
label: str = "persona"
class InternalTemplateBlockCreate(CreateBlock):
"""Used for Letta Cloud"""
base_template_id: str = Field(..., description="The id of the base template.")
template_id: str = Field(..., description="The id of the template.")
deployment_id: str = Field(..., description="The id of the deployment.")
entity_id: str = Field(..., description="The id of the entity within the template.")