Skip to content

Reports

Reports, sections and tasks, and their notebook, HTML and Word export.

lumen.ai.report

TContext = Mapping[str, Any] module-attribute

Action

Bases: ExecutableTask

An Action implements an execute method that performs some unit of work and optionally generates outputs to be rendered.

render_outputs = param.Boolean(default=True, doc='\n Whether the outputs should be rendered.') class-attribute instance-attribute

execute(context=None, **kwargs) async

servable(**kwargs)

Make the Action servable wrapped in a minimal Report UI.

show(**kwargs)

Show the Action wrapped in a minimal Report UI.

Actor

Bases: LLMUser

interface = param.ClassSelector(class_=ChatFeed, doc='\n The interface for the Coordinator to interact with.') class-attribute instance-attribute

respond(messages, context, **kwargs) abstractmethod async

Responds to the provided messages and context, returning a list of visual outputs and a ContextModel containing the context for subsequent Actors.

ActorTask

Bases: ExecutableTask

An ActorTask wraps an Actor in a way that allows persisting the message history, instructions, and context as well as making it possible to track its outputs.

actor = param.ClassSelector(class_=Actor) class-attribute instance-attribute

input_schema property

level = 3 class-attribute instance-attribute

output_schema property

editor(show_title=True)

Returns the editor for the tasks.

Arguments

show_title: bool Whether to show the title of the tasks.

Returns:

Type Description
The editor for the tasks.

prepare(context) async

ContextError

Bases: RuntimeError

Raised when context validation fails. Wraps a list of ValidationIssue objects and renders them as a readable ASCII tree in str().

issues = issues instance-attribute

title = title instance-attribute

summary()

Short one-line summary suitable for logs.

ContextProvider

Bases: Parameterized

ContextProvider describes the interface an Actor has to define to be invokable by other Actors, including its purpose so other actors can decide when to use it and the context values it requires and provides.

conditions = param.List(default=[], doc="\n Specific criteria that determine when this actor should be invoked.\n These conditions establish explicit relationships between actors in the system,\n defining the circumstances under which this actor becomes relevant.\n While 'purpose' describes what the actor does, conditions specify\n the precise situations that warrant its use.") class-attribute instance-attribute

exclusions = param.List(default=[], doc='\n List of context values that this actor should not be invoked with.\n This is useful for actors that are not relevant in certain contexts\n or for actors that should not be invoked in certain situations.') class-attribute instance-attribute

input_schema = ContextModel class-attribute instance-attribute

not_with = param.List(default=[], doc='\n List of actors that this actor should not be invoked with.') class-attribute instance-attribute

output_schema = ContextModel class-attribute instance-attribute

purpose = param.String(default='', doc="\n A descriptive statement of this actor's functionality and capabilities.\n Serves as a high-level explanation for other actors to understand\n what this actor does and when it might be useful to invoke it.") class-attribute instance-attribute

prepare(context) async

requirements(messages) async

ExecutableTask

Bases: Task, ContextProvider

level = 3 class-attribute instance-attribute

invalidate(keys, start=0, propagate=True)

Invalidates tasks and propagates context dependencies within a TaskGroup.

Parameters:

Name Type Description Default
keys Iterable[str]

A set of context keys that have been modified or invalidated by the user. These represent outputs whose dependent tasks should be re-evaluated.

 required
start int

Index of the task to start invalidating.

 0
propagate bool

Whether to propagate the invalidation to the parent task group.

 True

Returns:

Type Description
tuple[bool, set[str]]

A tuple (invalidated, keys) where: - invalidated is True if any tasks in this group (or nested groups) were marked invalid due to overlapping input dependencies. - keys is the full, cumulative set of invalidated context keys after propagating dependencies through all affected tasks.
Notes
  • If a task's input keys intersect with the provided invalidation keys, that task is considered stale and will be rerun.
  • Nested TaskGroups are traversed recursively, and their invalidations propagate upward to the parent group.

reset()

validate(context=None, available_types=None, path=None, raise_on_error=True)

Validate the task group and its subtasks.

Parameters:

Name Type Description Default
context TContext | None

The context to validate against, by default None

 None
available_types dict[str, Any] | None

Dictionary of available types for validation, by default None

 None
path str | None

Path to the current task group for error reporting, by default None

 None
raise_on_error bool

Whether to raise an error if validation issues are found, by default True

 True

Returns:

Type Description
tuple[list[ValidationIssue], dict[str, Any]]

A tuple containing: - List of validation issues found - Dictionary of output types from all tasks

FunctionTool

Bases: Tool

FunctionTool wraps arbitrary functions and makes them available as a tool for an LLM to call. It inspects the arguments of the function and generates a pydantic Model that the LLM will populate.

The function may also consume information in memory, e.g. the current table or pipeline by declaring the requires parameter.

The function may also return context to add to the current working memory by returning a dictionary and declaring the provides parameter. Any keys listed in the provides parameter will be copied into working memory.

conditions = param.List(default=[]) class-attribute instance-attribute

formatter = param.Parameter(default='{function}({arguments}) returned: {output}', doc="\n Formats the return value for inclusion in the global context.\n Accepts the 'function', 'arguments' and 'output' as formatting variables.") class-attribute instance-attribute

function = param.Callable(default=None, allow_refs=False, doc='\n The function to call.') class-attribute instance-attribute

inputs property

prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'FunctionTool' / 'main.jinja2'}}) class-attribute instance-attribute

provides = param.List(default=[], readonly=False, constant=True, doc='\n List of context values it provides to current working memory.') class-attribute instance-attribute

read_only = param.Boolean(default=False, doc='\n Whether calling the tool leaves all state unchanged. Identical calls\n to a read-only tool within one LLM request reuse the earlier result\n instead of running again.') class-attribute instance-attribute

render_output = param.Boolean(default=False, doc='\n Whether to render the tool output directly, even if it is not already a Lumen View or Panel Viewable.') class-attribute instance-attribute

requires = param.List(default=[], readonly=False, constant=True, doc='\n List of context values it requires to be in memory.') class-attribute instance-attribute

run_in_thread = param.Boolean(default=False, doc='\n Whether to run a synchronous function in a worker thread so that it\n does not block the event loop. Leave disabled for functions that\n interact with Panel objects, which may rely on running on the\n event loop thread.') class-attribute instance-attribute

respond(messages, context, **kwargs) async

LWW

Bases: TypedDict

A dictionary that overrides the last-write-wins (LWW) behavior.

Llm

Bases: Parameterized

Base class for LLM implementations with standardized client caching.

Subclasses MUST implement _create_base_client(). Subclasses MAY set _instructor_wrapper (default: "openai"). Subclasses MAY override _create_instructor_client() or _get_completion_method().

create_kwargs = param.Dict(default={'max_retries': 1}, doc='\n Additional keyword arguments to pass to the LLM provider\n when calling chat.completions.create. Defaults to no instructor retries\n since agents handle retries themselves.') class-attribute instance-attribute

display_name = param.String(default='', constant=True, doc='Display name for UI') class-attribute instance-attribute

interceptor = param.ClassSelector(default=None, class_=Interceptor, doc='\n Intercepter instance to intercept LLM calls, e.g. for logging.') class-attribute instance-attribute

logfire_tags = param.List(default=None, doc='\n Whether to log LLM calls and responses to logfire.\n If a list of tags is provided, those tags will be used for logging.\n Suppresses streaming responses if enabled since\n logfire does not track token usage on stream.') class-attribute instance-attribute

mode = param.Selector(default=(Mode.JSON_SCHEMA), objects=BASE_MODES, doc='\n The calling mode used by instructor to guide the LLM towards generating\n outputs matching the schema.') class-attribute instance-attribute

model_kwargs = param.Dict(default={}, doc='\n LLM model definitions indexed by type. Supported types include\n \'default\', \'reasoning\' and \'sql\'. Agents may pick which model to\n invoke for different reasons.\n\n An entry may declare an optional \'description\' (shown to the routing\n model so it can reason about what each option is for) and an optional\n \'routing\' key to opt in to model routing for that type: the routing\n model is invoked first to pick which entry to actually use for each\n call, e.g.\n {"default": {"model": "gpt-5.4-mini"},\n "edit": {"model": "gpt-5.2",\n "description": "Best for editing tables and visualizations",\n "routing": {"model": "nemotron-switchyard"}}}') class-attribute instance-attribute

select_models = param.List(default=[], constant=True, doc='Available models for selection dropdowns') class-attribute instance-attribute

spec_descriptions = param.Dict(default=SPEC_DESCRIPTIONS, doc='\n Mapping of spec key to human-readable description, used as a\n fallback in the routing prompt when ``model_kwargs`` entries\n do not declare their own ``description``. Seeded with the spec\n keys every provider shares and enriched by the UI layer from\n agent class metadata.') class-attribute instance-attribute

temperature = param.Number(default=0.7, bounds=(0, None), allow_None=True, constant=True, doc='\n The temperature to use for sampling. Set to None to omit the\n temperature from requests entirely, e.g. for models that reject it.') class-attribute instance-attribute

timeout = param.Number(default=120, bounds=(1, None), constant=True, doc='\n The timeout in seconds for API calls.') class-attribute instance-attribute

tools = param.List(default=[], doc='\n Default tools that are always available to this LLM instance.\n These are combined with any tools passed directly to invoke()\n or stream() calls. Accepts the same types as the tools argument\n on those methods (dicts, FunctionTool, or MCPTool instances).') class-attribute instance-attribute

usage = UsageCollector() instance-attribute

usage_pricing = param.Dict(default={}, doc='USD per million tokens by model:\n {"model-name": {"input": 1.0, "cached": 0.1, "cache_write": 1.25,\n "output": 4.0}}.\n Unknown models retain token counts but have no estimated cost.') class-attribute instance-attribute

capture_usage()

Capture provider usage for this LLM in the current async context.

initialize(log_level) async

invoke(messages, system='', response_model=None, allow_partial=False, model_spec='default', tools=None, submit_tool=None, **input_kwargs) async

Invokes the LLM and returns its response.

Arguments

messages: list[Message] A list of messages to feed to the LLM. system: str A system message to provide to the LLM. response_model: BaseModel | None A Pydantic model that the LLM should materialize. allow_partial: bool Whether to allow the LLM to only partially fill the provided response_model. tools: list[dict[str, Any] | FunctionTool | MCPTool] | None Tool definitions, FunctionTool, or MCPTool instances to pass through. When response_model is also given, the client runs tool calls in a loop until none are requested, then requests the structured response (tools may be unused). Tool errors, unknown tools and invalid arguments are returned to the model as tool results; max_tool_rounds (default 16) caps the rounds. submit_tool: SubmitTool | None Offer response_model as a tool so the model can answer inside the tool loop. Requires response_model. model: Literal['default' | 'reasoning' | 'sql'] The model as listed in the model_kwargs parameter to invoke to answer the query.

Returns:

Type Description
The completed response_model.

models()

Return the set of available model identifiers from this provider.

By default returns an empty set. Subclasses should override to query their provider's API for available models.

Returns:

Type Description
set[str]

Set of available model identifiers.

run_client(model_spec, messages, **kwargs) async

stream(messages, system='', response_model=None, field=None, model_spec='default', tools=None, **kwargs) async

Invokes the LLM and streams its response.

Arguments

messages: list[Message] A list of messages to feed to the LLM. system: str A system message to provide to the LLM. response_model: type[BaseModel] | None A Pydantic model that the LLM should materialize. field: str The field in the response_model to stream. tools: list[dict[str, Any] | FunctionTool | MCPTool] | None Tool definitions, FunctionTool, or MCPTool instances to pass through. model: Literal['default' | 'reasoning' | 'sql'] The model as listed in the model_kwargs parameter to invoke to answer the query.

Yields:

Type Description
The string or response_model field.

trace()

Capture model round trips and tool calls in the current async context.

warmup(model_kwargs) classmethod

Allows LLM provider to perform actions that ensure that the model(s) are ready to run, e.g. downloading the model files.

LumenEditor

Bases: Viewer

component = param.ClassSelector(class_=Component) class-attribute instance-attribute

editor = self._render_editor() instance-attribute

export_formats = ('yaml',) class-attribute instance-attribute

footer = param.List(default=[]) class-attribute instance-attribute

language = 'yaml' class-attribute instance-attribute

loading = param.Boolean() class-attribute instance-attribute

spec = param.String(allow_None=True) class-attribute instance-attribute

title = param.String(allow_None=True) class-attribute instance-attribute

view = ParamMethod(self.render, inplace=True, sizing_mode='stretch_width') instance-attribute

export(fmt)

render() async

render_context() async

render_controls(task, interface)

validate_spec(spec) classmethod

Message

Bases: TypedDict

content instance-attribute

name instance-attribute

role instance-attribute

tool_call_id instance-attribute

tool_calls instance-attribute

MissingContextError

Bases: Exception

Raise to indicate missing context for a query.

NullStep

status = None instance-attribute

stream(text, **kwargs)

Panel

Bases: View

Panel views provide a way to declaratively wrap a Panel component.

The Panel View is a very general purpose view that allows expressing arbitrary Panel objects as a specification. The Panel specification may be arbitrarily nested making it possible to specify entire layouts. Additionally the Panel specification also supports references, including standard source and variable references and a custom $data reference that inserts the current data of the View.

type: panel
  spec:
   type: panel.layout.Column
   objects:
     - type: pn.pane.Markdown
       object: '# My custom title'
     - type: pn.pane.DataFrame
       object: $data

object = Child() class-attribute instance-attribute

view_type = 'panel' class-attribute

from_spec(spec, source=None, filters=None, pipeline=None) classmethod

get_panel()

to_spec(context=None)

Report

Bases: TaskGroup

A Report is a TaskGroup consisting of a sequence of Section objects.

The Report UI renders buttons to execute, clear, export and configure the Report.

auto_execute = param.Boolean(default=False, doc='\n If True, automatically execute the report on initialization.') class-attribute instance-attribute

level = 1 class-attribute instance-attribute

from_views(views, title=None) classmethod

Assemble an exportable Report from prebuilt views without executing tasks.

Builds a task-less report whose export state is populated directly from already-rendered views, so to_html and to_notebook can produce a downloadable report without running an agent. The execute path is untouched; only views, the _view accordion and status are set, and the empty task list keeps the export guards satisfied.

Parameters:

Name Type Description Default
views list

The prebuilt viewable outputs to include, in order.

 required
title str

The report title, used for the exported document.

 None

to_docx()

Returns the Word (.docx) representation of the report as bytes.

to_html()

Returns the HTML representation of the report.

to_notebook()

Returns the notebook representation of the report.

Section

Bases: TaskGroup

A Section is a TaskGroup representing a sequence of related tasks.

level = 2 class-attribute instance-attribute

Task

Bases: Viewer

A Task defines a single unit of work that can be executed and rendered.

abort_on_error = param.Boolean(default=True, doc='\n If True, the report will abort if an error occurs.') class-attribute instance-attribute

context = param.Dict(default={}, allow_refs=True, doc='\n The context to pass to the task.') class-attribute instance-attribute

history = param.List(doc='\n Conversation history to include as context for the task.') class-attribute instance-attribute

include_in_export = param.Boolean(default=True, doc="\n Whether this task's outputs are included when the report is exported.") class-attribute instance-attribute

instruction = param.String(default='', doc='\n The instruction to give to the task.') class-attribute instance-attribute

interface = param.ClassSelector(class_=ChatFeed, doc='\n The chat interface to use for the task.') class-attribute instance-attribute

llm = param.ClassSelector(class_=Llm, doc='\n The LLM to use for the task.') class-attribute instance-attribute

out_context = param.Dict(default={}, doc='\n The context generated by the task.') class-attribute instance-attribute

parent = param.Parameter(doc='\n The parent task group.') class-attribute instance-attribute

running = param.Boolean(doc='\n Whether the task is currently running.') class-attribute instance-attribute

status = param.Selector(objects=['idle', 'running', 'success', 'error', 'cancelled'], default='idle', doc='\n The current status of the task.') class-attribute instance-attribute

steps_layout = param.ClassSelector(default=None, class_=(ListLike, NamedListLike), allow_None=True, doc='\n The layout progress updates will be streamed to.') class-attribute instance-attribute

title = param.String(doc='\n The title of the task.') class-attribute instance-attribute

views = param.List(allow_refs=True, doc='\n The generated viewable outputs of the task.') class-attribute instance-attribute

cleanup()

Cleanup when the task is not longer needed.

Distinct from reset which only resets the outputs.

editor(show_title=True)

Returns the editor for the task.

Arguments

show_title: bool Whether to show the title of the task.

Returns:

Type Description
The editor for the task.

execute(context=None, **kwargs) async

Executes the task.

Arguments

**kwargs: dict Additional keyword arguments to pass to the task.

Returns:

Type Description
The outputs of the task.

prepare(context) async

reset()

Resets the view, removing generated outputs.

TaskGroup

Bases: Task

level = 3 class-attribute instance-attribute

append(task)

Appends a task to the collection.

Arguments

task: Task | Actor The task to append.

cleanup()

editor(show_title=True)

Returns the editor for the tasks.

Arguments

show_title: bool Whether to show the title of the tasks.

Returns:

Type Description
The editor for the tasks.

insert(index, task)

Inserts a task at the given index.

Arguments

index: int The index to insert the task at. task: Task | Actor The task to insert.

invalidate(keys, start=0, propagate=True)

Invalidates tasks and propagates context dependencies within a TaskGroup.

Parameters:

Name Type Description Default
keys Iterable[str]

A set of context keys that have been modified or invalidated by the user. These represent outputs whose dependent tasks should be re-evaluated.

 required
start int

Index of the task to start invalidating.

 0
propagate bool

Whether to propagate the invalidation to the parent task group.

 True

Returns:

Type Description
tuple[bool, set[str]]

A tuple (invalidated, keys) where: - invalidated is True if any tasks in this group (or nested groups) were marked invalid due to overlapping input dependencies. - keys is the full, cumulative set of invalidated context keys after propagating dependencies through all affected tasks.
Notes
  • If a task's input keys intersect with the provided invalidation keys, that task is considered stale and will be rerun.
  • Nested TaskGroups are traversed recursively, and their invalidations propagate upward to the parent group.

merge(other)

Merges another task group into the current task group.

Parameters:

Name Type Description Default
other TaskGroup

The other task group to merge with.

 required

Returns:

Name Type Description
self TaskGroup

The current task group.

prepare(context) async

remove(task)

reset(start=0)

Resets the view, removing generated outputs.

validate(context=None, available_types=None, path=None, raise_on_error=True)

Validate the task group and its subtasks.

Parameters:

Name Type Description Default
context TContext | None

The context to validate against, by default None

 None
available_types dict[str, Any] | None

Dictionary of available types for validation, by default None

 None
path str | None

Path to the current task group for error reporting, by default None

 None
raise_on_error bool

Whether to raise an error if validation issues are found, by default True

 True

Returns:

Type Description
tuple[list[ValidationIssue], dict[str, Any]]

A tuple containing: - List of validation issues found - Dictionary of output types from all tasks

Tool

Bases: Actor, ContextProvider

A Tool can be invoked by another Actor to provide additional context or respond to a question. Unlike an Agent they never interact with or on behalf of a user directly.

always_use = param.Boolean(default=False, doc='\n Whether to always use this tool, even if it is not explicitly\n required by the current context.') class-attribute instance-attribute

conditions = param.List(default=['Always requires a supporting agent to interpret results']) class-attribute instance-attribute

applies(context) async classmethod

Additional checks to determine if the tool should be used.

prepare(context) async

Prepare the tool with the initial context. Called once when the tool is first initialized.

sync(context) async

Allows the tool to update when the provided context changes. Subclasses should override this to handle context updates.

ValidationIssue dataclass

actual instance-attribute

expected instance-attribute

key instance-attribute

message instance-attribute

path instance-attribute

severity = 'error' class-attribute instance-attribute

View

Bases: MultiTypeComponent, Viewer

View components provide a visual representation for the data returned by a :class:lumen.source.base.Source or :class:lumen.pipeline.Pipeline.

The View must return a Panel object or an object that can be rendered by Panel. The base class provides methods which query the the provided :class:lumen.pipeline.Pipeline.

Subclasses should use these methods to query the data and return a Viewable Panel object in the get_panel method.

control_panel property

controls = param.List(default=[], doc='\n Parameters that should be exposed as widgets in the UI.') class-attribute instance-attribute

download = param.ClassSelector(class_=Download, default=(Download()), doc='\n The download objects determines whether and how the source tables\n can be downloaded.') class-attribute instance-attribute

field = param.Selector(doc='The field being visualized.') class-attribute instance-attribute

kwargs = {k: v for k, v in (params.items()) if k not in self.param} instance-attribute

limit = param.Integer(default=None, bounds=(0, None), doc='\n Limits the number of rows that are rendered.') class-attribute instance-attribute

loading_indicator = param.Boolean(default=True, constant=True, doc='\n Whether to display a loading indicator on the View when the\n Pipeline is refreshing the data.') class-attribute instance-attribute

panel property

pipeline = param.ClassSelector(class_=Pipeline, doc='\n The data pipeline that drives the View.') class-attribute instance-attribute

rerender = param.Event(default=False, doc='\n An event that is triggered whenever the View requests a re-render.') class-attribute instance-attribute

selection_group = param.String(default=None, doc='\n Declares a selection group the plot is part of. This feature\n requires the separate HoloViews library.') class-attribute instance-attribute

title = param.String(default=None, doc='\n The title of the view.') class-attribute instance-attribute

view_type = None class-attribute

from_spec(spec, source=None, filters=None, pipeline=None) classmethod

Resolves a View specification given the schema of the Source it will be filtering on.

Parameters:

Name Type Description Default
spec dict[str, Any] | str

Specification declared as a dictionary of parameter values.

 required
source

The Source object containing the tables the View renders.

 None
filters

A list of Filter objects which provide query values for the Source.

 None
pipeline

The Lumen pipeline driving this View. Must not be supplied if the spec contains a pipeline definition or reference.

 None

Returns:

Type Description
The resolved View object.

get_data()

Queries the Source for the specified table applying any filters and transformations specified on the View. Unlike get_value this should be used when multiple return values are expected.

Returns:

Type Description
DataFrame

The queried table after filtering and transformations are applied.

get_panel()

Constructs and returns a Panel object which will represent a view of the queried table.

Returns:

Type Description
Viewable

A Panel Viewable object representing a current representation of the queried table.

get_value(field=None)

Queries the Source for the data associated with a particular field applying any filters and transformations specified on the View. Unlike get_data this method returns a single scalar value associated with the field and should therefore only be used if only a single.

Parameters:

Name Type Description Default
field str | None

The field from the table to return; if None uses field defined on the View.

 None

Returns:

Type Description
object

A single scalar value representing the current value of the queried field.

to_spec(context=None)

Exports the full specification to reconstruct this component.

Parameters:

Name Type Description Default
context dict[str, Any] | None

Context contains the specification of all previously serialized components, e.g. to allow resolving of references.

 None

Returns:

Type Description
Declarative specification of this component.

update(*events, invalidate_cache=True)

Triggers an update in the View.

Parameters:

Name Type Description Default
events Event

param events that may trigger an update.

 ()
invalidate_cache bool

Whether to clear the View's cache.

 True

_export_checkbox(task, status_source, align='center')

Checkbox controlling whether a task's outputs are kept when the report is exported. status_source supplies the status it tracks, so it only appears once that has something to export.

_export_header(section, status_source, variant='h3')

Heading for a section, with the checkbox that keeps or discards it.

collect_task_outputs(task)

Returns {key: type} for what this task guarantees to add/enrich in context, based on task.output_schema (a TypedDict schema).

content_to_text(content)

Extract the text portion from a message content field.

Handles both plain string content and multimodal content lists (e.g. containing both text strings and image objects). Returns just the concatenated text parts.

Parameters:

Name Type Description Default
content str | list

The message content — either a plain string or a list of mixed content items (strings, Image objects, etc.).

 required

Returns:

Type Description
str

The extracted text content.

docx_add_chart(doc, editor)

Embed a chart editor as a PNG image; return True if one was added.

docx_add_markdown(doc, text)

Render basic markdown (headings, bullets, bold) into a docx document.

docx_add_table(doc, df, max_rows=50)

Render a DataFrame as a native Word table (header + capped rows).

extract_block_source(template_path, block_name, relative_to=PROMPTS_DIR)

format_output(output)

get_block_names(template_path, relative_to=PROMPTS_DIR)

input_dependency_keys(schema)

Keys that, if invalidated, require rerunning a task with this inputs schema. Includes the field names themselves PLUS any ('accumulate','') sources.

make_md_cell(text)

make_preamble(preamble, extensions=None, title=None)

merge_contexts(schema, contexts, *, lww_keys=None)

Merge a list of context dicts according to the schema's Annotated metadata. - Accumulated fields: extend/append from 'from_key' occurrences across contexts - Regular fields: last-write-wins (LWW) unless overridden by lww_keys

Parameters:

Name Type Description Default
schema type[TypedDict]

The schema to merge the contexts according to.

 required
contexts list[Mapping[str, Any]]

The list of contexts to merge.

 required

Returns:

Type Description
dict[str, Any]

The merged context.

set_content_text(new_text, content)

Return a new content value with the text part replaced by new_text, preserving any non-text items (e.g. images) in a multimodal list.

Parameters:

Name Type Description Default
new_text str

The replacement text.

 required
content str | list

The original message content — either a plain string or a multimodal list of strings and other objects.

 required

Returns:

Type Description
str | list

new_text when content is a plain string, or [new_text, *non_text_items] when non-text items exist, or new_text when the list contained only strings.

try_import(module_name, load=True)

Import and return a module, or None if it is unavailable.

With load=False the module is returned only if it has already been imported, a cheap check that never triggers an import; this lets a hot path ask whether an optional dependency is in play without paying to import it. With load=True (the default) the module is imported on demand.

validate_task_inputs(task, value_ctx, available_types, path)

Validates that task.input_schema are satisfied by value_ctx or prior available_types.

validate_taskgroup_exclusions(group, *, path='TaskGroup')

Validate that within this TaskGroup there are no tasks that exclude each other. A conflict occurs if task_i.not_with contains task_j's CLASS NAME, or vice versa.

Recurses into nested TaskGroups, but only checks conflicts within each group, not across siblings or parents.

wrap_logfire_on_method(cls, method_name)

Automatically adds logfire wrapping to a method.

write_notebook(cells)