@chat.function
@chat.function — the typed decorator that turns a Python coroutine into an LLM-callable Webbee tool, with auto-validated params, action types, and multi-step turns.
@chat.function is the most important decorator in the SDK. It's how you tell Webbee "here is a tool you can call when users speak". This page covers every option, every guarantee, every gotcha.
What it does
When you write:
@chat.function(
"send_message",
description="Send an email message.",
)
async def send_message(ctx, params: SendMessageParams) -> ActionResult:
...Three things happen at build time:
- The SDK extracts a JSON schema from
SendMessageParamsand embeds it in your manifest. - The web-kernel registers
send_messageas a tool the intent classifier can choose. - The Pydantic model becomes the runtime gatekeeper — every call is auto-validated.
At runtime, when the LLM picks your tool:
LLM: "I'll call send_message with {to: ['sarah@x.com'], subject: 'Q3'}"
↓
SDK: Pydantic.SendMessageParams(**args) → ValidationError? retry up to 2× with prose feedback
↓
SDK: ctx pre-flight (UNKNOWN_SUB_FUNCTION, FABRICATED_ID_SHAPE)
↓
Your code: send_message(ctx, params)
↓
SDK: audit chokepoint records (user_id, tenant_id, tool_name, action_type, status)All decorator parameters
Prop
Type
The Pydantic params model — the right way
from pydantic import BaseModel, Field
# ✅ MODULE-SCOPE — this is enforced by V17 validator
class SendMessageParams(BaseModel):
to: list[str] = Field(description="Recipients — list of email addresses")
subject: str = Field(description="Email subject line")
body: str = Field(description="Plain-text or HTML body")
cc: list[str] | None = Field(None, description="Optional CC recipients")
reply_to_thread_id: str | None = Field(
None,
description="Optional thread UUID to reply within an existing conversation",
)
@chat.function(
"send_message",
description="Send an email message via the user's connected mail account.",
action_type="write",
)
async def send_message(ctx, params: SendMessageParams) -> ActionResult:
...Don't define Pydantic models inside functions
Function-local Pydantic models silently disable the auto-detection used by the SDK. The retry feedback loop won't run. V17 catches this at validate-time:
# ❌ DON'T
async def my_handler(ctx, **kwargs):
class Params(BaseModel): ... # function-local — V17 failsWhat every Field needs
| Field attribute | Why | Example |
|---|---|---|
description= | LLM reads this to fill the param correctly | "ISO datetime, e.g. 2026-06-15T09:00:00" |
| Default value | Marks the field as optional in the JSON schema | Field(None, ...) or Field([], ...) |
| Type annotation | Drives validation + JSON schema | list[str], str | None, Literal['a','b'] |
The fewer the required fields, the better — but each required field MUST have a description so the LLM doesn't hallucinate.
Action types
action_type is a federal classification that affects what the platform does before your handler runs:
read
Pure read. No side effects. No confirmation. The classifier may call this freely.
write
Side-effecting. Audit row written. May appear in confirmation cards as part of a multi-step turn.
destructive
Irreversible (deletes, sends, charges). ALWAYS prompts the user with a confirmation card before firing.
@chat.function("read_calendar", description="Read the user's calendar.", action_type="read") # No confirmation
@chat.function("send_email", description="Send an email.", action_type="write") # Audit only
@chat.function("delete_project", description="Permanently delete a project.", action_type="destructive") # Confirmation cardThe Pre-Authorized Action Execution flow
When a destructive tool fires — or a write/destructive step inside a multi-step turn — the web-kernel:
- Intercepts the call before it executes.
- Renders a card to the user — "This will delete project 'Acme'. Confirm?"
- The user's "yes" hits a typed-iterate path that runs your handler with the exact same args, byte-for-byte. No LLM rerun.
- After execution, the audit row reflects user-confirmed acceptance.
The platform guarantees that confirming runs exactly what you were shown — the same action, with the same arguments — and it's all free for you.
Return shapes
Your handler returns an ActionResult — the universal return type for @chat.function. The federal typed-return contract (validator V18) requires the handler return annotation to resolve to ActionResult (or a subclass); a bare -> dict annotation fails the publish/validate gate. The structured payload you put inside ActionResult.success(...) is an SDL value — an sdl.Entity for a single record, or a concrete sdl.EntityList[T] for a list — declared via data_model= (V23/V24). Build results with the .success() / .error() factory methods — never construct directly.
from imperal_sdk import ActionResult, sdl
# An SDL entity declared at module scope, passed as data_model=Message
class Message(sdl.Entity, sdl.Correspondents, sdl.MessageState):
pass
# Success — summary is the user-facing chat line; data is the SDL entity
return ActionResult.success(
Message(id="abc", title="Q3", recipients_to=["sarah@company.com"], delivery_state="sent"),
summary="Email sent to sarah@company.com.",
)
# Success with a list — return a concrete sdl.EntityList[T]; the brain can pass its fields into a later call
class Task(sdl.Entity, sdl.Completable):
pass
class TaskList(sdl.EntityList[Task]):
pass
return ActionResult.success(
TaskList(items=[Task(id="t1", title="Ship SDL")], total=3),
summary="Found 3 tasks.",
)
# Success with inline UI — attach a UINode tree for rich chat rendering
return ActionResult.success(
Message(id="d1", title="Q3"),
summary="Drafted. Send?",
ui=preview_card,
)
# Error — rendered as a chat error; set retryable when the user can try again
return ActionResult.error("No such project.", retryable=False)The federal-clean way: always provide a summary on success (it is what the user sees in chat) and return the SDL entity/list as the payload. Declare a matching data_model= so the platform reads the entity's roles and the brain can pass its fields into a later call. See the SDL reference for the facet/role library.
Multi-step turns
Webbee's brain iterates. It calls a tool, the real result folds back into its context, it reads that result, and it decides the next call — repeating for a bounded handful of steps until the turn is done. "Do two things from one message" works by this iteration, not by any plan drawn up in advance.
Take "check my unread mail and create a note for the most important one":
- The brain calls
list_unreadand gets a realEntityListof messages back in its context. - It reads those actual messages — subjects, senders, ids — and picks the important one.
- It calls
create_note, writing the arguments from what it just saw.
Nothing is scripted ahead of time; each call is chosen from the concrete results of the last one. Keeping chain_callable=True (the default) lets the platform dispatch to your tool as one of those follow-up calls; set it False only if your tool cannot be safely used mid-turn (rare).
This is exactly why the typed return contract matters. When your read tool returns a real sdl.Entity / sdl.EntityList[T] (declared via data_model=), the brain sees exact field values — a message's real id, its real subject — and passes those exact values into the next call, instead of paraphrasing them out of chat prose.
Typed dispatch
When Webbee already knows the exact tool and typed arguments for a step, the platform dispatches directly via Pydantic — no extra LLM round inside your extension. This closed a class of bugs where BYOLLM routers stochastically split tool calls across rounds.
The Pydantic feedback loop
When the LLM sends arguments that fail validation, the SDK does not immediately fail. Instead:
1. LLM emits create_task({description: "Q3 report"}) ← missing required 'title'
2. SDK: ValidationError("title: required field is missing")
3. SDK: format_pydantic_for_llm(e) → structured prose
4. SDK: re-prompt LLM with the prose feedback
5. LLM emits create_task({title: "Q3 report", description: "..."})
6. SDK: validates — OK — your handler runsBounded retry: at most 2 retries per tool_use, beyond that → standard VALIDATION_MISSING_FIELD failure.
This closed ~75% of arg-quality hallucinations in production. See the Pydantic feedback loop reference for invariants and observability.
Target-id resolution — id_projection
When Webbee resolves a target entity for your tool, the web-kernel needs to know which Pydantic field receives the resolved id. It resolves that field structurally — from your explicit id_projection, or otherwise from a *_id field in your params model. The kernel does not guess the field from the tool name (the old verb-prefix heuristic was purged on 2026-06-03).
id_projection is the structural, SDL-aligned, federal-clean way to declare it. Declare it whenever the target field is not a single obvious <noun>_id:
@chat.function(
"delete_notes_from_folder",
description="Delete all notes inside a folder.",
action_type="destructive",
id_projection="folder_id", # ← structural target-id field; never name-guessed
)
async def delete_notes_from_folder(ctx, params) -> ActionResult: ...When you need this
Whenever the target field isn't an obvious <noun>_id matching the entity — e.g. compound names (delete_notes_from_folder → folder_id) or a write that targets a parent (create_task → project_id). Declaring it is always safe.
Common patterns
Each pattern declares its SDL return type (from imperal_sdk import sdl) and points data_model= at it. Pydantic (BaseModel, Field) is still how you type the input *Params.
class Note(sdl.Entity, sdl.Excerptable):
pass
class NoteList(sdl.EntityList[Note]):
pass
class SearchParams(BaseModel):
query: str = Field(description="Search term")
limit: int = Field(10, description="Max results to return")
@chat.function(
"search_notes",
description="Search the user's notes by full-text query.",
action_type="read",
data_model=NoteList, # ← required for reads (V23)
)
async def search_notes(ctx, params: SearchParams) -> ActionResult:
rows = await ctx.http.get(f"/notes/search?q={params.query}&limit={params.limit}")
items = [Note(id=r["id"], title=r["title"], excerpt=r.get("snippet")) for r in rows]
return ActionResult.success(
NoteList(items=items, total=len(items)),
summary=f"Found {len(items)} match(es).",
)class Note(sdl.Entity, sdl.Bodied):
folder_id: str | None = None
class CreateNoteParams(BaseModel):
title: str = Field(description="Note title")
content_text: str = Field(description="The full note content — write the actual content, not a placeholder")
folder_id: str | None = Field(None, description="Optional UUID of folder to put it in")
@chat.function(
"create_note",
description="Create a new note. The content_text MUST contain the actual user-requested content.",
action_type="write",
data_model=Note, # ← recommended for writes (V24): return the created entity
)
async def create_note(ctx, params: CreateNoteParams) -> ActionResult:
note = await ctx.http.post("/notes", json=params.model_dump(exclude_none=True))
return ActionResult.success(
Note(id=note["id"], title=note["title"], url=note["panel_url"], folder_id=note.get("folder_id")),
summary=f"Created note: {note['title']}",
)class FolderDeleted(sdl.Entity):
pass
class DeleteFolderParams(BaseModel):
folder_id: str = Field(description="UUID of the folder to delete")
@chat.function(
"delete_folder",
description="Permanently delete a folder and all its notes.",
action_type="destructive",
id_projection="folder_id",
data_model=FolderDeleted,
)
async def delete_folder(ctx, params: DeleteFolderParams) -> ActionResult:
# Confirmation card already shown before this runs
await ctx.http.delete(f"/folders/{params.folder_id}")
return ActionResult.success(
FolderDeleted(id=params.folder_id, title="Folder"),
summary="Folder deleted.",
)class Message(sdl.Entity, sdl.MessageState):
pass
class MessageList(sdl.EntityList[Message]):
pass
class GetUnreadParams(BaseModel):
limit: int = Field(20, description="Max unread to return")
@chat.function(
"get_unread",
description="Return the user's unread emails. Useful as the first step in a multi-step turn like 'summarize my unread'.",
action_type="read",
chain_callable=True,
data_model=MessageList,
)
async def get_unread(ctx, params: GetUnreadParams) -> ActionResult:
msgs = await ctx.http.get(f"/mail/unread?limit={params.limit}")
items = [Message(id=m["id"], title=m["subject"], is_read=False) for m in msgs]
return ActionResult.success(
MessageList(items=items, total=len(items)), # ← the brain can pass its fields into a later call
summary=f"{len(items)} unread.",
)Things that bite
What's next
SDL reference
sdl.Entity, sdl.EntityList[T], facets and roles — the required shape for every data_model= return.
Skeletons
Live data feeds — how Webbee always knows your state.
Panels
UI surfaces in the Imperal Panel app.
Web-kernel context (ctx)
The full ctx object — user, tenant, api, secrets, history, lang.
Manifest reference
What ends up in imperal.json — every field.
Extensions
What an Imperal Cloud extension is — its files, lifecycle, and capabilities; the Webbee superpower you build, publish, and ship to users as a Python package.
Skeletons
Skeletons — live ambient state and entity feeds that keep Webbee aware of user state, enabling instant routing without exploratory tool calls.