API Reference

This section provides a complete API reference for the HAP package.

haive.hap.models

haive.hap.models.context

class haive.hap.models.context.HAPContext(*, engine=None, engines=<factory>, execution_path=<factory>, agent_metadata=<factory>, graph_context=<factory>, legacy_inputs=<factory>, legacy_outputs=<factory>, legacy_state=<factory>, legacy_meta=<factory>)[source]

Bases: StateSchema

HAP execution context inheriting from real Haive StateSchema.

Parameters:
agent_metadata: Dict[str, Any]
execution_path: List[str]
graph_context: Dict[str, Any]
property inputs: Dict[str, Any]

Backward compatibility for inputs.

legacy_inputs: Dict[str, Any]
legacy_meta: Dict[str, Any]
legacy_outputs: Dict[str, Any]
legacy_state: Dict[str, Any]
property meta: Dict[str, Any]

Backward compatibility for meta.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property outputs: Dict[str, Any]

Backward compatibility for outputs.

property state: Dict[str, Any]

Backward compatibility for state.

class haive.hap.models.context.HAPContext(*, engine=None, engines=<factory>, execution_path=<factory>, agent_metadata=<factory>, graph_context=<factory>, legacy_inputs=<factory>, legacy_outputs=<factory>, legacy_state=<factory>, legacy_meta=<factory>)[source]

HAP execution context inheriting from real Haive StateSchema.

Parameters:
classmethod __get_pydantic_json_schema__(core_schema, handler, /)

Hook into generating the model’s JSON schema.

Parameters:

  • core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.

  • handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

classmethod __pydantic_init_subclass__(**kwargs)

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after the class is actually fully initialized. In particular, attributes like model_fields will be present when this is called.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by pydantic.

Parameters:

**kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by pydantic.

Return type:

None

classmethod as_table()

Create a rich table representation of the schema.

Returns:

Rich Table object

Return type:

Table

classmethod compare_with(other, title=None)

Compare this schema with another in a side-by-side display.

Parameters:

  • other (type[StateSchema]) – Other schema to compare with

  • title (str | None) – Optional title for the comparison

Return type:

None

classmethod construct(_fields_set=None, **values)
Parameters:
Return type:

Self

classmethod create_input_schema(engine_name=None, name=None)

Alias for derive_input_schema for backward compatibility.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for input validation

Return type:

type[BaseModel]

classmethod create_output_schema(engine_name=None, name=None)

Alias for derive_output_schema for backward compatibility.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for output validation

Return type:

type[BaseModel]

classmethod derive_input_schema(engine_name=None, name=None)

Derive an input schema for the given engine from this state schema.

This method intelligently selects the appropriate base class for the derived schema, using prebuilt states (MessagesState, ToolState) when appropriate instead of just creating a generic BaseModel.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target (default: all inputs)

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for input validation, potentially inheriting from MessagesState or ToolState for better compatibility

Return type:

type[BaseModel]

classmethod derive_output_schema(engine_name=None, name=None)

Derive an output schema for the given engine from this state schema.

This method intelligently selects the appropriate base class for the derived schema, using prebuilt states (MessagesState, ToolState) when appropriate instead of just creating a generic BaseModel.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target (default: all outputs)

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for output validation, potentially inheriting from MessagesState or ToolState for better compatibility

Return type:

type[BaseModel]

classmethod display_code(title=None)

Display Python code representation of the schema.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

classmethod display_schema(title=None)

Display schema information in a rich format.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

classmethod display_table()

Display schema as a table.

Return type:

None

classmethod extract_values(state, keys=None)

Class method to extract values from a state object or dictionary.

Parameters:

  • state (StateSchema | dict[str, Any]) – State object or dictionary to extract values from

  • keys (list[str] | dict[str, str] | None) – Can be: - List[str]: List of field names to extract - Dict[str, str]: Mapping of output keys to state field names - None: Extract all fields

Returns:

Dictionary containing the requested values

Return type:

dict[str, Any]

classmethod from_dict(data)

Create a state from a dictionary.

Parameters:

data (FieldMapping) – Dictionary with field values

Returns:

New StateSchema instance

Return type:

Self

classmethod from_json(json_str)

Create state from JSON string.

Parameters:

json_str (str) – JSON string to parse

Returns:

New StateSchema instance

Return type:

StateSchema

classmethod from_orm(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod from_partial_dict(data)

Create a state from a partial dictionary, filling in defaults.

Parameters:

data (dict[str, Any]) – Partial dictionary with field values

Returns:

New StateSchema instance with defaults applied

Return type:

StateSchema

classmethod from_runnable_config(config)

Extract state from a RunnableConfig.

Parameters:

config (RunnableConfig) – RunnableConfig to extract from

Returns:

StateSchema instance or None if no state found

Return type:

StateSchema | None

classmethod from_snapshot(snapshot)

Create a state from a LangGraph StateSnapshot.

Parameters:

snapshot (Any) – StateSnapshot from LangGraph

Returns:

New StateSchema instance

Return type:

StateSchema

classmethod get_all_class_engines()

Get all class-level engines.

Returns:

Dictionary of all engines

Return type:

dict[str, Any]

classmethod get_class_engine(name)

Get a class-level engine by name.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

classmethod get_structured_model(model_name)

Get a structured output model class by name.

Parameters:

model_name (str) – Name of the structured model

Returns:

Model class if found, None otherwise

Return type:

type[BaseModel] | None

classmethod is_shared(field_name)

Check if a field is shared with parent graphs.

Parameters:

field_name (str) – Field name to check

Returns:

True if field is shared, False otherwise

Return type:

bool

classmethod list_structured_models()

List all structured output models in this schema.

Returns:

List of structured model names

Return type:

list[str]

classmethod manager()

Get a manager for this schema (shorthand for to_manager()).

Returns:

StateSchemaManager instance

Return type:

StateSchemaManager

classmethod model_construct(_fields_set=None, **values)

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note

model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

  • _fields_set (set[str] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.

  • values (Any) – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

Return type:

Self

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation')

Generates a JSON schema for a model class.

Parameters:

  • by_alias (bool) – Whether to use attribute aliases or not.

  • ref_template (str) – The reference template.

  • schema_generator (type[GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

  • mode (Literal['validation', 'serialization']) – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

Return type:

dict[str, Any]

classmethod model_parametrized_name(params)

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:

params (tuple[type[Any], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.

Returns:

String representing the new class where params are passed to cls as type variables.

Raises:

TypeError – Raised when trying to generate concrete names for non-generic models.

Return type:

str

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

  • force (bool) – Whether to force the rebuilding of the model schema, defaults to False.

  • raise_errors (bool) – Whether to raise errors, defaults to True.

  • _parent_namespace_depth (int) – The depth level of the parent namespace, defaults to 2.

  • _types_namespace (MappingNamespace | None) – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Return type:

bool | None

classmethod model_validate(obj, *, strict=None, from_attributes=None, context=None, by_alias=None, by_name=None)

Validate a pydantic model instance.

Parameters:

  • obj (Any) – The object to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • from_attributes (bool | None) – Whether to extract data from object attributes.

  • context (Any | None) – Additional context to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

Return type:

Self

classmethod model_validate_json(json_data, *, strict=None, context=None, by_alias=None, by_name=None)
!!! abstract “Usage Documentation”

[JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

  • json_data (str | bytes | bytearray) – The JSON data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Return type:

Self

classmethod model_validate_strings(obj, *, strict=None, context=None, by_alias=None, by_name=None)

Validate the given object with string data against the Pydantic model.

Parameters:

  • obj (Any) – The object containing string data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Return type:

Self

classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • path (str | Path)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • b (str | bytes)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias=True, ref_template='#/$defs/{model}')
Parameters:

  • by_alias (bool)

  • ref_template (str)

Return type:

Dict[str, Any]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)
Parameters:

  • by_alias (bool)

  • ref_template (str)

  • dumps_kwargs (Any)

Return type:

str

classmethod shared_fields()

Get the list of fields shared with parent graphs.

Returns:

List of shared field names

Return type:

list[str]

classmethod to_manager(name=None)

Convert schema class to a StateSchemaManager for further manipulation.

Parameters:

name (str | None) – Optional name for the resulting manager

Returns:

StateSchemaManager instance

Return type:

StateSchemaManager

classmethod to_python_code()

Convert schema to Python code representation.

Returns:

String containing Python code representation

Return type:

str

classmethod update_forward_refs(**localns)
Parameters:

localns (Any)

Return type:

None

classmethod validate(value)
Parameters:

value (Any)

Return type:

Self

classmethod validate_engine(v)

Handle both serialized dict and actual Engine instances.

This validator allows the engine field to accept both: - Actual Engine instances (for runtime use) - Serialized dicts (for state passing between agents)

This prevents the “Can’t instantiate abstract class Engine” error when deserializing state in multi-agent systems.

Return type:

Any

classmethod validate_engines(v)

Handle both serialized dicts and actual Engine instances in engines dict.

Similar to validate_engine but for the engines dictionary. Each value can be either a serialized dict or an actual Engine instance.

Return type:

Any

classmethod with_shared_fields(fields)

Create a copy of this schema with specified shared fields.

Parameters:

fields (list[str]) – List of field names to be marked as shared

Returns:

New StateSchema subclass with updated shared fields

Return type:

type[StateSchema]

__copy__()

Returns a shallow copy of the model.

Return type:

Self

__deepcopy__(memo=None)

Returns a deep copy of the model.

Parameters:

memo (dict[int, Any] | None)

Return type:

Self

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

Return type:

None

__iter__()

So dict(model) works.

Return type:

Generator[tuple[str, Any], None, None]

__pretty__(fmt, **kwargs)

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:
Return type:

Generator[Any, None, None]

__repr_name__()

Name of the instance’s class, used in __repr__.

Return type:

str

__repr_recursion__(object)

Returns the string representation of a recursive object.

Parameters:

object (Any)

Return type:

str

__rich_repr__()

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:

RichReprResult

add_engine(name, engine)

Add an engine to the engines registry.

Parameters:

  • name (str) – Name/key for the engine

  • engine (Engine) – Engine instance to add

Return type:

None

add_message(message)

Add a single message to the messages field.

Parameters:

message (BaseMessage) – BaseMessage to add

Returns:

Self for chaining

Return type:

StateSchema

add_messages(new_messages)

Add multiple messages to the messages field.

Parameters:

new_messages (list[BaseMessage]) – List of messages to add

Returns:

Self for chaining

Return type:

StateSchema

apply_reducers(other)

Update state applying reducer functions where defined.

This method processes updates with special handling for fields that have reducer functions defined.

Parameters:

other (dict[str, Any] | StateSchema) – Dictionary or StateSchema with update values

Returns:

Self for chaining

Return type:

StateSchema

clear_messages()

Clear all messages in the messages field.

Returns:

Self for chaining

Return type:

StateSchema

combine_with(other)

Combine this state with another, applying reducers for shared fields.

This is more sophisticated than update() or apply_reducers() as it properly handles StateSchema-specific metadata and shared fields.

Parameters:

other (StateSchema | dict[str, Any]) – Other state to combine with

Returns:

New combined state instance

Return type:

StateSchema

copy(**updates)

Create a copy of this state, optionally with updates.

Parameters:

**updates – Field values to update in the copy

Returns:

New StateSchema instance

Return type:

StateSchema

deep_copy()

Create a deep copy of this state object.

Returns:

New StateSchema instance with deep-copied values

Return type:

StateSchema

dict(**kwargs)

Backwards compatibility alias for model_dump.

Parameters:

**kwargs – Keyword arguments for model_dump

Returns:

Dictionary representation of the state

Return type:

dict[str, Any]

differences_from(other)

Compare this state with another and return differences.

Parameters:

other (StateSchema | dict[str, Any]) – Other state to compare with

Returns:

Dictionary mapping field names to (self_value, other_value) tuples

Return type:

dict[str, tuple[Any, Any]]

get(key, default=None)

Safely get a field value with a default.

Parameters:

  • key (str) – Field name to get

  • default (Any) – Default value if field doesn’t exist

Returns:

Field value or default

Return type:

Any

get_all_instance_engines()

Get all engines from both instance and class level.

Returns:

Dictionary mapping engine names to engine instances

Return type:

dict[str, Any]

get_engine(name)

Get an engine by name from any engine fields.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

get_engines()

Get all engines in this state.

Returns:

Dictionary mapping engine names to engine instances

Return type:

dict[str, Any]

get_instance_engine(name)

Get an engine from instance or class level.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

get_last_message()

Get the last message in the messages field.

Returns:

Last message or None if no messages exist

Return type:

BaseMessage | None

get_state_values(keys=None)

Extract specified state values into a dictionary.

Parameters:

keys (list[str] | dict[str, str] | None) – Can be: - List[str]: List of field names to extract - Dict[str, str]: Mapping of output keys to state field names - None: Extract all fields

Returns:

Dictionary containing the requested state values

Return type:

dict[str, Any]

has_engine(name)

Check if an engine exists in this state.

Parameters:

name (str) – Name of the engine to check

Returns:

True if engine exists, False otherwise

Return type:

bool

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)
Parameters:
Return type:

str

list_engines()

Get list of all engine names.

Returns:

List of engine names

Return type:

list[str]

merge_engine_output(engine_name, output, apply_reducers=True)

Merge output from an engine into this state.

Parameters:

  • engine_name (str) – Name of the engine

  • output (dict[str, Any]) – Output data from the engine

  • apply_reducers (bool) – Whether to apply reducers during merge

Returns:

Self for chaining

Return type:

StateSchema

merge_messages(new_messages)

Merge new messages with existing messages using appropriate reducer.

Parameters:

new_messages (list[BaseMessage]) – New messages to add

Returns:

Self for chaining

Return type:

StateSchema

model_copy(*, update=None, deep=False)
!!! abstract “Usage Documentation”

[model_copy](../concepts/serialization.md#model_copy)

Returns a copy of the model.

!!! note

The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

  • update (Mapping[str, Any] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.

  • deep (bool) – Set to True to make a deep copy of the model.

Returns:

New model instance.

Return type:

Self

model_dump(**kwargs)

Override model_dump to exclude internal fields and handle special types.

Parameters:

**kwargs (Any) – Keyword arguments for model_dump

Returns:

Dictionary representation of the state

Return type:

FieldMapping

model_dump_json(*, indent=None, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump_json](../concepts/serialization.md#modelmodel_dump_json)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

  • indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to exclude from the JSON output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to serialize using field aliases.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A JSON string representation of the model.

Return type:

str

model_post_init(_StateSchema__context)

Sync engines from class level to instance level after initialization.

This ensures that engines stored at the class level (via SchemaComposer) are available on state instances.

Parameters:

_StateSchema__context (Any)

Return type:

None

patch(update_data, apply_reducers=True)

Update specific fields in the state.

Parameters:

  • update_data (dict[str, Any]) – Dictionary of field updates

  • apply_reducers (bool) – Whether to apply reducer functions

Returns:

Self for chaining

Return type:

StateSchema

prepare_for_engine(engine_name)

Prepare state data for a specific engine.

Extracts only fields that are inputs for the specified engine.

Parameters:

engine_name (str) – Name of the engine to prepare for

Returns:

Dictionary with engine-specific inputs

Return type:

dict[str, Any]

pretty_print(title=None)

Print state with rich formatting for easy inspection.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

remove_engine(name)

Remove an engine from the registry.

Parameters:

name (str) – Name of the engine to remove

Returns:

True if engine was removed, False if not found

Return type:

bool

setup_engines_and_tools()

Setup engines and sync their tools, structured output models, and add engine to state.

This validator runs after the model is created and: 1. Finds all engine fields in the state 2. Syncs engine to main engine field and engines dict 3. Syncs tools from engine to state tools field 4. Syncs structured output models 5. Sets up parent-child relationships for nested state schemas

Return type:

Self

sync_engine_fields()

Sync between engine and engines dict for backward compatibility.

This validator ensures that: 1. If ‘engine’ is set, it’s available in engines dict 2. If engines dict has items but no engine, set main engine 3. Both access patterns work seamlessly

Return type:

Self

to_command(goto=None, graph=None)

Convert state to a Command object for LangGraph control flow.

Parameters:

  • goto (str | None) – Optional next node to go to

  • graph (str | None) – Optional graph to target (None for current, PARENT for parent)

Returns:

Command object with state update

Return type:

Any

to_dict()

Convert the state to a clean dictionary.

Returns:

Dictionary representation of the state

Return type:

FieldMapping

to_json()

Convert state to JSON string.

Returns:

JSON string representation of the state

Return type:

str

to_runnable_config(thread_id=None, **kwargs)

Convert state to a RunnableConfig.

Parameters:

  • thread_id (str | None) – Optional thread ID for the configuration

  • **kwargs – Additional configuration parameters

Returns:

RunnableConfig containing state data

Return type:

RunnableConfig

update(other)

Update the state with values from another state or dictionary.

This method performs a simple update without applying reducers.

Parameters:

other (dict[str, Any] | StateSchema) – Dictionary or StateSchema with update values

Returns:

Self for chaining

Return type:

StateSchema

agent_metadata: Dict[str, Any]
engine: TEngine | None
engines: builtins.dict[str, Engine]
execution_path: List[str]
graph_context: Dict[str, Any]
property inputs: Dict[str, Any]

Backward compatibility for inputs.

legacy_inputs: Dict[str, Any]
legacy_meta: Dict[str, Any]
legacy_outputs: Dict[str, Any]
legacy_state: Dict[str, Any]
property llm: Engine | None

Convenience property to access the LLM engine.

property main_engine: Engine | None

Convenience property to access the main engine.

property meta: Dict[str, Any]

Backward compatibility for meta.

model_computed_fields = {}
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property model_extra: dict[str, Any] | None

Get extra fields set during validation.

Returns:

A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'agent_metadata': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Metadata about agents in execution'), 'engine': FieldInfo(annotation=Union[TypeVar, NoneType], required=False, default=None, description='Optional main/primary engine for convenience'), 'engines': FieldInfo(annotation=dict[str, Engine], required=False, default_factory=dict, description='Engine registry for this state - supports easy addition'), 'execution_path': FieldInfo(annotation=List[str], required=False, default_factory=list, description='Path of agents/nodes executed'), 'graph_context': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Graph-level execution context'), 'legacy_inputs': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_meta': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_outputs': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_state': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict)}
property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,

i.e. that were not filled from defaults.

property outputs: Dict[str, Any]

Backward compatibility for outputs.

property state: Dict[str, Any]

Backward compatibility for state.

haive.hap.models.graph

haive.hap.models.graph.AgentGraph

alias of HAPGraph

haive.hap.models.graph.AgentNode

alias of HAPNode

class haive.hap.models.graph.HAPGraph(*, nodes=<factory>, entry_node='')[source]

Bases: BaseModel

HAP graph with agent orchestration capabilities.

Parameters:
add_agent_node(node_id, agent, next_nodes=None)[source]

Add an agent as a node to the graph.

Parameters:

  • node_id (str)

  • agent (Agent)

  • next_nodes (List[str])

Return type:

HAPNode

add_entrypoint_node(node_id, entrypoint, next_nodes=None)[source]

Add a node by entrypoint string.

Parameters:
Return type:

HAPNode

async execute(initial_context)[source]

Execute the entire graph.

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

topological_order()[source]

Get topological ordering of nodes.

Return type:

List[str]

entry_node: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodes: Dict[str, HAPNode]
class haive.hap.models.graph.HAPNode(*, id, agent_entrypoint, agent_instance=None, next_nodes=<factory>)[source]

Bases: BaseModel

HAP node that can contain an agent.

Parameters:

  • id (str)

  • agent_entrypoint (str)

  • agent_instance (Agent | None)

  • next_nodes (List[str])

async execute(context)[source]

Execute this node’s agent.

Parameters:

context (HAPContext)

Return type:

HAPContext

load_agent()[source]

Load agent from entrypoint if not already loaded.

Return type:

Agent

agent_entrypoint: str
agent_instance: Agent | None
id: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

next_nodes: List[str]
class haive.hap.models.graph.HAPGraph(*, nodes=<factory>, entry_node='')[source]

HAP graph with agent orchestration capabilities.

Parameters:
classmethod __get_pydantic_json_schema__(core_schema, handler, /)

Hook into generating the model’s JSON schema.

Parameters:

  • core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.

  • handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

classmethod __pydantic_init_subclass__(**kwargs)

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after the class is actually fully initialized. In particular, attributes like model_fields will be present when this is called.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by pydantic.

Parameters:

**kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by pydantic.

Return type:

None

classmethod construct(_fields_set=None, **values)
Parameters:
Return type:

Self

classmethod from_orm(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod model_construct(_fields_set=None, **values)

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note

model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

  • _fields_set (set[str] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.

  • values (Any) – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

Return type:

Self

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation')

Generates a JSON schema for a model class.

Parameters:

  • by_alias (bool) – Whether to use attribute aliases or not.

  • ref_template (str) – The reference template.

  • schema_generator (type[GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

  • mode (Literal['validation', 'serialization']) – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

Return type:

dict[str, Any]

classmethod model_parametrized_name(params)

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:

params (tuple[type[Any], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.

Returns:

String representing the new class where params are passed to cls as type variables.

Raises:

TypeError – Raised when trying to generate concrete names for non-generic models.

Return type:

str

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

  • force (bool) – Whether to force the rebuilding of the model schema, defaults to False.

  • raise_errors (bool) – Whether to raise errors, defaults to True.

  • _parent_namespace_depth (int) – The depth level of the parent namespace, defaults to 2.

  • _types_namespace (MappingNamespace | None) – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Return type:

bool | None

classmethod model_validate(obj, *, strict=None, from_attributes=None, context=None, by_alias=None, by_name=None)

Validate a pydantic model instance.

Parameters:

  • obj (Any) – The object to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • from_attributes (bool | None) – Whether to extract data from object attributes.

  • context (Any | None) – Additional context to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

Return type:

Self

classmethod model_validate_json(json_data, *, strict=None, context=None, by_alias=None, by_name=None)
!!! abstract “Usage Documentation”

[JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

  • json_data (str | bytes | bytearray) – The JSON data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Return type:

Self

classmethod model_validate_strings(obj, *, strict=None, context=None, by_alias=None, by_name=None)

Validate the given object with string data against the Pydantic model.

Parameters:

  • obj (Any) – The object containing string data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Return type:

Self

classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • path (str | Path)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • b (str | bytes)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias=True, ref_template='#/$defs/{model}')
Parameters:

  • by_alias (bool)

  • ref_template (str)

Return type:

Dict[str, Any]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)
Parameters:

  • by_alias (bool)

  • ref_template (str)

  • dumps_kwargs (Any)

Return type:

str

classmethod update_forward_refs(**localns)
Parameters:

localns (Any)

Return type:

None

classmethod validate(value)
Parameters:

value (Any)

Return type:

Self

__copy__()

Returns a shallow copy of the model.

Return type:

Self

__deepcopy__(memo=None)

Returns a deep copy of the model.

Parameters:

memo (dict[int, Any] | None)

Return type:

Self

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

Return type:

None

__iter__()

So dict(model) works.

Return type:

Generator[tuple[str, Any], None, None]

__pretty__(fmt, **kwargs)

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:
Return type:

Generator[Any, None, None]

__repr_name__()

Name of the instance’s class, used in __repr__.

Return type:

str

__repr_recursion__(object)

Returns the string representation of a recursive object.

Parameters:

object (Any)

Return type:

str

__rich_repr__()

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:

RichReprResult

add_agent_node(node_id, agent, next_nodes=None)[source]

Add an agent as a node to the graph.

Parameters:

  • node_id (str)

  • agent (Agent)

  • next_nodes (List[str])

Return type:

HAPNode

add_entrypoint_node(node_id, entrypoint, next_nodes=None)[source]

Add a node by entrypoint string.

Parameters:
Return type:

HAPNode

copy(*, include=None, exclude=None, update=None, deep=False)

Returns a copy of the model.

!!! warning “Deprecated”

This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

`python {test="skip" lint="skip"} data = self.model_dump(include=include, exclude=exclude, round_trip=True) data = {**data, **(update or {})} copied = self.model_validate(data) `

Parameters:

  • include (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to include in the copied model.

  • exclude (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to exclude in the copied model.

  • update (Dict[str, Any] | None) – Optional dictionary of field-value pairs to override field values in the copied model.

  • deep (bool) – If True, the values of fields that are Pydantic models will be deep-copied.

Returns:

A copy of the model with included, excluded and updated fields as specified.

Return type:

Self

dict(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False)
Parameters:
Return type:

Dict[str, Any]

async execute(initial_context)[source]

Execute the entire graph.

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)
Parameters:
Return type:

str

model_copy(*, update=None, deep=False)
!!! abstract “Usage Documentation”

[model_copy](../concepts/serialization.md#model_copy)

Returns a copy of the model.

!!! note

The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

  • update (Mapping[str, Any] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.

  • deep (bool) – Set to True to make a deep copy of the model.

Returns:

New model instance.

Return type:

Self

model_dump(*, mode='python', include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump](../concepts/serialization.md#modelmodel_dump)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

  • mode (Literal['json', 'python'] | str) – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to include in the output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to exclude from the output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to use the field’s alias in the dictionary key if defined.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A dictionary representation of the model.

Return type:

dict[str, Any]

model_dump_json(*, indent=None, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump_json](../concepts/serialization.md#modelmodel_dump_json)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

  • indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to exclude from the JSON output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to serialize using field aliases.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A JSON string representation of the model.

Return type:

str

model_post_init(context, /)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Parameters:

context (Any)

Return type:

None

topological_order()[source]

Get topological ordering of nodes.

Return type:

List[str]

entry_node: str
model_computed_fields = {}
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property model_extra: dict[str, Any] | None

Get extra fields set during validation.

Returns:

A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'entry_node': FieldInfo(annotation=str, required=False, default='', description='Entry point node ID'), 'nodes': FieldInfo(annotation=Dict[str, HAPNode], required=False, default_factory=dict, description='HAP-specific nodes')}
property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,

i.e. that were not filled from defaults.

nodes: Dict[str, HAPNode]
class haive.hap.models.graph.HAPNode(*, id, agent_entrypoint, agent_instance=None, next_nodes=<factory>)[source]

HAP node that can contain an agent.

Parameters:

  • id (str)

  • agent_entrypoint (str)

  • agent_instance (Agent | None)

  • next_nodes (List[str])

classmethod __get_pydantic_json_schema__(core_schema, handler, /)

Hook into generating the model’s JSON schema.

Parameters:

  • core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.

  • handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

classmethod __pydantic_init_subclass__(**kwargs)

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after the class is actually fully initialized. In particular, attributes like model_fields will be present when this is called.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by pydantic.

Parameters:

**kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by pydantic.

Return type:

None

classmethod construct(_fields_set=None, **values)
Parameters:
Return type:

Self

classmethod from_orm(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod model_construct(_fields_set=None, **values)

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note

model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

  • _fields_set (set[str] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.

  • values (Any) – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

Return type:

Self

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation')

Generates a JSON schema for a model class.

Parameters:

  • by_alias (bool) – Whether to use attribute aliases or not.

  • ref_template (str) – The reference template.

  • schema_generator (type[GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

  • mode (Literal['validation', 'serialization']) – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

Return type:

dict[str, Any]

classmethod model_parametrized_name(params)

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:

params (tuple[type[Any], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.

Returns:

String representing the new class where params are passed to cls as type variables.

Raises:

TypeError – Raised when trying to generate concrete names for non-generic models.

Return type:

str

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

  • force (bool) – Whether to force the rebuilding of the model schema, defaults to False.

  • raise_errors (bool) – Whether to raise errors, defaults to True.

  • _parent_namespace_depth (int) – The depth level of the parent namespace, defaults to 2.

  • _types_namespace (MappingNamespace | None) – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Return type:

bool | None

classmethod model_validate(obj, *, strict=None, from_attributes=None, context=None, by_alias=None, by_name=None)

Validate a pydantic model instance.

Parameters:

  • obj (Any) – The object to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • from_attributes (bool | None) – Whether to extract data from object attributes.

  • context (Any | None) – Additional context to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

Return type:

Self

classmethod model_validate_json(json_data, *, strict=None, context=None, by_alias=None, by_name=None)
!!! abstract “Usage Documentation”

[JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

  • json_data (str | bytes | bytearray) – The JSON data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Return type:

Self

classmethod model_validate_strings(obj, *, strict=None, context=None, by_alias=None, by_name=None)

Validate the given object with string data against the Pydantic model.

Parameters:

  • obj (Any) – The object containing string data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Return type:

Self

classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • path (str | Path)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • b (str | bytes)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias=True, ref_template='#/$defs/{model}')
Parameters:

  • by_alias (bool)

  • ref_template (str)

Return type:

Dict[str, Any]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)
Parameters:

  • by_alias (bool)

  • ref_template (str)

  • dumps_kwargs (Any)

Return type:

str

classmethod update_forward_refs(**localns)
Parameters:

localns (Any)

Return type:

None

classmethod validate(value)
Parameters:

value (Any)

Return type:

Self

__copy__()

Returns a shallow copy of the model.

Return type:

Self

__deepcopy__(memo=None)

Returns a deep copy of the model.

Parameters:

memo (dict[int, Any] | None)

Return type:

Self

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

Return type:

None

__iter__()

So dict(model) works.

Return type:

Generator[tuple[str, Any], None, None]

__pretty__(fmt, **kwargs)

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:
Return type:

Generator[Any, None, None]

__repr_name__()

Name of the instance’s class, used in __repr__.

Return type:

str

__repr_recursion__(object)

Returns the string representation of a recursive object.

Parameters:

object (Any)

Return type:

str

__rich_repr__()

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:

RichReprResult

copy(*, include=None, exclude=None, update=None, deep=False)

Returns a copy of the model.

!!! warning “Deprecated”

This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

`python {test="skip" lint="skip"} data = self.model_dump(include=include, exclude=exclude, round_trip=True) data = {**data, **(update or {})} copied = self.model_validate(data) `

Parameters:

  • include (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to include in the copied model.

  • exclude (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to exclude in the copied model.

  • update (Dict[str, Any] | None) – Optional dictionary of field-value pairs to override field values in the copied model.

  • deep (bool) – If True, the values of fields that are Pydantic models will be deep-copied.

Returns:

A copy of the model with included, excluded and updated fields as specified.

Return type:

Self

dict(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False)
Parameters:
Return type:

Dict[str, Any]

async execute(context)[source]

Execute this node’s agent.

Parameters:

context (HAPContext)

Return type:

HAPContext

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)
Parameters:
Return type:

str

load_agent()[source]

Load agent from entrypoint if not already loaded.

Return type:

Agent

model_copy(*, update=None, deep=False)
!!! abstract “Usage Documentation”

[model_copy](../concepts/serialization.md#model_copy)

Returns a copy of the model.

!!! note

The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

  • update (Mapping[str, Any] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.

  • deep (bool) – Set to True to make a deep copy of the model.

Returns:

New model instance.

Return type:

Self

model_dump(*, mode='python', include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump](../concepts/serialization.md#modelmodel_dump)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

  • mode (Literal['json', 'python'] | str) – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to include in the output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to exclude from the output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to use the field’s alias in the dictionary key if defined.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A dictionary representation of the model.

Return type:

dict[str, Any]

model_dump_json(*, indent=None, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump_json](../concepts/serialization.md#modelmodel_dump_json)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

  • indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to exclude from the JSON output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to serialize using field aliases.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A JSON string representation of the model.

Return type:

str

model_post_init(context, /)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Parameters:

context (Any)

Return type:

None

agent_entrypoint: str
agent_instance: Agent | None
id: str
model_computed_fields = {}
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property model_extra: dict[str, Any] | None

Get extra fields set during validation.

Returns:

A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'agent_entrypoint': FieldInfo(annotation=str, required=True, description='Module:Class entrypoint for agent'), 'agent_instance': FieldInfo(annotation=Union[Agent, NoneType], required=False, default=None, description='Loaded agent instance', exclude=True), 'id': FieldInfo(annotation=str, required=True, description='Node identifier'), 'next_nodes': FieldInfo(annotation=List[str], required=False, default_factory=list, description='Next node IDs in execution order')}
property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,

i.e. that were not filled from defaults.

next_nodes: List[str]

haive.hap.server

haive.hap.server.runtime

class haive.hap.server.runtime.HAPRuntime(graph)[source]

Bases: object

Execute an HAP graph with proper error handling and Haive integration.

Parameters:

graph (HAPGraph)

__init__(graph)[source]
Parameters:

graph (HAPGraph)

async run(initial_context)[source]

Execute the graph asynchronously.

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

run_sync(initial_context)[source]

Execute the graph synchronously (for backward compatibility).

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

class haive.hap.server.runtime.HAPRuntime(graph)[source]

Execute an HAP graph with proper error handling and Haive integration.

Parameters:

graph (HAPGraph)

__init__(graph)[source]
Parameters:

graph (HAPGraph)

_load_agent(entrypoint)[source]

Load agent from entrypoint with proper error handling.

Parameters:

entrypoint (str)

Return type:

Agent

async run(initial_context)[source]

Execute the graph asynchronously.

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

run_sync(initial_context)[source]

Execute the graph synchronously (for backward compatibility).

Parameters:

initial_context (Dict[str, Any])

Return type:

HAPContext

haive.hap.hap

Note: HAP (Haive Agent Protocol) uses JSON-RPC 2.0 for agent communication.

haive.hap.hap.context

class haive.hap.server.runtime.HAPContext(*, engine=None, engines=<factory>, execution_path=<factory>, agent_metadata=<factory>, graph_context=<factory>, legacy_inputs=<factory>, legacy_outputs=<factory>, legacy_state=<factory>, legacy_meta=<factory>)[source]

HAP execution context inheriting from real Haive StateSchema.

Parameters:
classmethod __get_pydantic_json_schema__(core_schema, handler, /)

Hook into generating the model’s JSON schema.

Parameters:

  • core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.

  • handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

classmethod __pydantic_init_subclass__(**kwargs)

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after the class is actually fully initialized. In particular, attributes like model_fields will be present when this is called.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by pydantic.

Parameters:

**kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by pydantic.

Return type:

None

classmethod as_table()

Create a rich table representation of the schema.

Returns:

Rich Table object

Return type:

Table

classmethod compare_with(other, title=None)

Compare this schema with another in a side-by-side display.

Parameters:

  • other (type[StateSchema]) – Other schema to compare with

  • title (str | None) – Optional title for the comparison

Return type:

None

classmethod construct(_fields_set=None, **values)
Parameters:
Return type:

Self

classmethod create_input_schema(engine_name=None, name=None)

Alias for derive_input_schema for backward compatibility.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for input validation

Return type:

type[BaseModel]

classmethod create_output_schema(engine_name=None, name=None)

Alias for derive_output_schema for backward compatibility.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for output validation

Return type:

type[BaseModel]

classmethod derive_input_schema(engine_name=None, name=None)

Derive an input schema for the given engine from this state schema.

This method intelligently selects the appropriate base class for the derived schema, using prebuilt states (MessagesState, ToolState) when appropriate instead of just creating a generic BaseModel.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target (default: all inputs)

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for input validation, potentially inheriting from MessagesState or ToolState for better compatibility

Return type:

type[BaseModel]

classmethod derive_output_schema(engine_name=None, name=None)

Derive an output schema for the given engine from this state schema.

This method intelligently selects the appropriate base class for the derived schema, using prebuilt states (MessagesState, ToolState) when appropriate instead of just creating a generic BaseModel.

Parameters:

  • engine_name (str | None) – Optional name of the engine to target (default: all outputs)

  • name (str | None) – Optional name for the schema class

Returns:

A BaseModel subclass for output validation, potentially inheriting from MessagesState or ToolState for better compatibility

Return type:

type[BaseModel]

classmethod display_code(title=None)

Display Python code representation of the schema.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

classmethod display_schema(title=None)

Display schema information in a rich format.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

classmethod display_table()

Display schema as a table.

Return type:

None

classmethod extract_values(state, keys=None)

Class method to extract values from a state object or dictionary.

Parameters:

  • state (StateSchema | dict[str, Any]) – State object or dictionary to extract values from

  • keys (list[str] | dict[str, str] | None) – Can be: - List[str]: List of field names to extract - Dict[str, str]: Mapping of output keys to state field names - None: Extract all fields

Returns:

Dictionary containing the requested values

Return type:

dict[str, Any]

classmethod from_dict(data)

Create a state from a dictionary.

Parameters:

data (FieldMapping) – Dictionary with field values

Returns:

New StateSchema instance

Return type:

Self

classmethod from_json(json_str)

Create state from JSON string.

Parameters:

json_str (str) – JSON string to parse

Returns:

New StateSchema instance

Return type:

StateSchema

classmethod from_orm(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod from_partial_dict(data)

Create a state from a partial dictionary, filling in defaults.

Parameters:

data (dict[str, Any]) – Partial dictionary with field values

Returns:

New StateSchema instance with defaults applied

Return type:

StateSchema

classmethod from_runnable_config(config)

Extract state from a RunnableConfig.

Parameters:

config (RunnableConfig) – RunnableConfig to extract from

Returns:

StateSchema instance or None if no state found

Return type:

StateSchema | None

classmethod from_snapshot(snapshot)

Create a state from a LangGraph StateSnapshot.

Parameters:

snapshot (Any) – StateSnapshot from LangGraph

Returns:

New StateSchema instance

Return type:

StateSchema

classmethod get_all_class_engines()

Get all class-level engines.

Returns:

Dictionary of all engines

Return type:

dict[str, Any]

classmethod get_class_engine(name)

Get a class-level engine by name.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

classmethod get_structured_model(model_name)

Get a structured output model class by name.

Parameters:

model_name (str) – Name of the structured model

Returns:

Model class if found, None otherwise

Return type:

type[BaseModel] | None

classmethod is_shared(field_name)

Check if a field is shared with parent graphs.

Parameters:

field_name (str) – Field name to check

Returns:

True if field is shared, False otherwise

Return type:

bool

classmethod list_structured_models()

List all structured output models in this schema.

Returns:

List of structured model names

Return type:

list[str]

classmethod manager()

Get a manager for this schema (shorthand for to_manager()).

Returns:

StateSchemaManager instance

Return type:

StateSchemaManager

classmethod model_construct(_fields_set=None, **values)

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note

model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

  • _fields_set (set[str] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.

  • values (Any) – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

Return type:

Self

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation')

Generates a JSON schema for a model class.

Parameters:

  • by_alias (bool) – Whether to use attribute aliases or not.

  • ref_template (str) – The reference template.

  • schema_generator (type[GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

  • mode (Literal['validation', 'serialization']) – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

Return type:

dict[str, Any]

classmethod model_parametrized_name(params)

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:

params (tuple[type[Any], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.

Returns:

String representing the new class where params are passed to cls as type variables.

Raises:

TypeError – Raised when trying to generate concrete names for non-generic models.

Return type:

str

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

  • force (bool) – Whether to force the rebuilding of the model schema, defaults to False.

  • raise_errors (bool) – Whether to raise errors, defaults to True.

  • _parent_namespace_depth (int) – The depth level of the parent namespace, defaults to 2.

  • _types_namespace (MappingNamespace | None) – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Return type:

bool | None

classmethod model_validate(obj, *, strict=None, from_attributes=None, context=None, by_alias=None, by_name=None)

Validate a pydantic model instance.

Parameters:

  • obj (Any) – The object to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • from_attributes (bool | None) – Whether to extract data from object attributes.

  • context (Any | None) – Additional context to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

Return type:

Self

classmethod model_validate_json(json_data, *, strict=None, context=None, by_alias=None, by_name=None)
!!! abstract “Usage Documentation”

[JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

  • json_data (str | bytes | bytearray) – The JSON data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Return type:

Self

classmethod model_validate_strings(obj, *, strict=None, context=None, by_alias=None, by_name=None)

Validate the given object with string data against the Pydantic model.

Parameters:

  • obj (Any) – The object containing string data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Return type:

Self

classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • path (str | Path)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj)
Parameters:

obj (Any)

Return type:

Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Parameters:

  • b (str | bytes)

  • content_type (str | None)

  • encoding (str)

  • proto (DeprecatedParseProtocol | None)

  • allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias=True, ref_template='#/$defs/{model}')
Parameters:

  • by_alias (bool)

  • ref_template (str)

Return type:

Dict[str, Any]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)
Parameters:

  • by_alias (bool)

  • ref_template (str)

  • dumps_kwargs (Any)

Return type:

str

classmethod shared_fields()

Get the list of fields shared with parent graphs.

Returns:

List of shared field names

Return type:

list[str]

classmethod to_manager(name=None)

Convert schema class to a StateSchemaManager for further manipulation.

Parameters:

name (str | None) – Optional name for the resulting manager

Returns:

StateSchemaManager instance

Return type:

StateSchemaManager

classmethod to_python_code()

Convert schema to Python code representation.

Returns:

String containing Python code representation

Return type:

str

classmethod update_forward_refs(**localns)
Parameters:

localns (Any)

Return type:

None

classmethod validate(value)
Parameters:

value (Any)

Return type:

Self

classmethod validate_engine(v)

Handle both serialized dict and actual Engine instances.

This validator allows the engine field to accept both: - Actual Engine instances (for runtime use) - Serialized dicts (for state passing between agents)

This prevents the “Can’t instantiate abstract class Engine” error when deserializing state in multi-agent systems.

Return type:

Any

classmethod validate_engines(v)

Handle both serialized dicts and actual Engine instances in engines dict.

Similar to validate_engine but for the engines dictionary. Each value can be either a serialized dict or an actual Engine instance.

Return type:

Any

classmethod with_shared_fields(fields)

Create a copy of this schema with specified shared fields.

Parameters:

fields (list[str]) – List of field names to be marked as shared

Returns:

New StateSchema subclass with updated shared fields

Return type:

type[StateSchema]

__copy__()

Returns a shallow copy of the model.

Return type:

Self

__deepcopy__(memo=None)

Returns a deep copy of the model.

Parameters:

memo (dict[int, Any] | None)

Return type:

Self

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

Return type:

None

__iter__()

So dict(model) works.

Return type:

Generator[tuple[str, Any], None, None]

__pretty__(fmt, **kwargs)

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:
Return type:

Generator[Any, None, None]

__repr_name__()

Name of the instance’s class, used in __repr__.

Return type:

str

__repr_recursion__(object)

Returns the string representation of a recursive object.

Parameters:

object (Any)

Return type:

str

__rich_repr__()

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:

RichReprResult

add_engine(name, engine)

Add an engine to the engines registry.

Parameters:

  • name (str) – Name/key for the engine

  • engine (Engine) – Engine instance to add

Return type:

None

add_message(message)

Add a single message to the messages field.

Parameters:

message (BaseMessage) – BaseMessage to add

Returns:

Self for chaining

Return type:

StateSchema

add_messages(new_messages)

Add multiple messages to the messages field.

Parameters:

new_messages (list[BaseMessage]) – List of messages to add

Returns:

Self for chaining

Return type:

StateSchema

apply_reducers(other)

Update state applying reducer functions where defined.

This method processes updates with special handling for fields that have reducer functions defined.

Parameters:

other (dict[str, Any] | StateSchema) – Dictionary or StateSchema with update values

Returns:

Self for chaining

Return type:

StateSchema

clear_messages()

Clear all messages in the messages field.

Returns:

Self for chaining

Return type:

StateSchema

combine_with(other)

Combine this state with another, applying reducers for shared fields.

This is more sophisticated than update() or apply_reducers() as it properly handles StateSchema-specific metadata and shared fields.

Parameters:

other (StateSchema | dict[str, Any]) – Other state to combine with

Returns:

New combined state instance

Return type:

StateSchema

copy(**updates)

Create a copy of this state, optionally with updates.

Parameters:

**updates – Field values to update in the copy

Returns:

New StateSchema instance

Return type:

StateSchema

deep_copy()

Create a deep copy of this state object.

Returns:

New StateSchema instance with deep-copied values

Return type:

StateSchema

dict(**kwargs)

Backwards compatibility alias for model_dump.

Parameters:

**kwargs – Keyword arguments for model_dump

Returns:

Dictionary representation of the state

Return type:

dict[str, Any]

differences_from(other)

Compare this state with another and return differences.

Parameters:

other (StateSchema | dict[str, Any]) – Other state to compare with

Returns:

Dictionary mapping field names to (self_value, other_value) tuples

Return type:

dict[str, tuple[Any, Any]]

get(key, default=None)

Safely get a field value with a default.

Parameters:

  • key (str) – Field name to get

  • default (Any) – Default value if field doesn’t exist

Returns:

Field value or default

Return type:

Any

get_all_instance_engines()

Get all engines from both instance and class level.

Returns:

Dictionary mapping engine names to engine instances

Return type:

dict[str, Any]

get_engine(name)

Get an engine by name from any engine fields.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

get_engines()

Get all engines in this state.

Returns:

Dictionary mapping engine names to engine instances

Return type:

dict[str, Any]

get_instance_engine(name)

Get an engine from instance or class level.

Parameters:

name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

get_last_message()

Get the last message in the messages field.

Returns:

Last message or None if no messages exist

Return type:

BaseMessage | None

get_state_values(keys=None)

Extract specified state values into a dictionary.

Parameters:

keys (list[str] | dict[str, str] | None) – Can be: - List[str]: List of field names to extract - Dict[str, str]: Mapping of output keys to state field names - None: Extract all fields

Returns:

Dictionary containing the requested state values

Return type:

dict[str, Any]

has_engine(name)

Check if an engine exists in this state.

Parameters:

name (str) – Name of the engine to check

Returns:

True if engine exists, False otherwise

Return type:

bool

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)
Parameters:
Return type:

str

list_engines()

Get list of all engine names.

Returns:

List of engine names

Return type:

list[str]

merge_engine_output(engine_name, output, apply_reducers=True)

Merge output from an engine into this state.

Parameters:

  • engine_name (str) – Name of the engine

  • output (dict[str, Any]) – Output data from the engine

  • apply_reducers (bool) – Whether to apply reducers during merge

Returns:

Self for chaining

Return type:

StateSchema

merge_messages(new_messages)

Merge new messages with existing messages using appropriate reducer.

Parameters:

new_messages (list[BaseMessage]) – New messages to add

Returns:

Self for chaining

Return type:

StateSchema

model_copy(*, update=None, deep=False)
!!! abstract “Usage Documentation”

[model_copy](../concepts/serialization.md#model_copy)

Returns a copy of the model.

!!! note

The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

  • update (Mapping[str, Any] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.

  • deep (bool) – Set to True to make a deep copy of the model.

Returns:

New model instance.

Return type:

Self

model_dump(**kwargs)

Override model_dump to exclude internal fields and handle special types.

Parameters:

**kwargs (Any) – Keyword arguments for model_dump

Returns:

Dictionary representation of the state

Return type:

FieldMapping

model_dump_json(*, indent=None, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump_json](../concepts/serialization.md#modelmodel_dump_json)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

  • indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.

  • include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.

  • exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to exclude from the JSON output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to serialize using field aliases.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (bool | Literal['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Callable[[Any], Any] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A JSON string representation of the model.

Return type:

str

model_post_init(_StateSchema__context)

Sync engines from class level to instance level after initialization.

This ensures that engines stored at the class level (via SchemaComposer) are available on state instances.

Parameters:

_StateSchema__context (Any)

Return type:

None

patch(update_data, apply_reducers=True)

Update specific fields in the state.

Parameters:

  • update_data (dict[str, Any]) – Dictionary of field updates

  • apply_reducers (bool) – Whether to apply reducer functions

Returns:

Self for chaining

Return type:

StateSchema

prepare_for_engine(engine_name)

Prepare state data for a specific engine.

Extracts only fields that are inputs for the specified engine.

Parameters:

engine_name (str) – Name of the engine to prepare for

Returns:

Dictionary with engine-specific inputs

Return type:

dict[str, Any]

pretty_print(title=None)

Print state with rich formatting for easy inspection.

Parameters:

title (str | None) – Optional title for the display

Return type:

None

remove_engine(name)

Remove an engine from the registry.

Parameters:

name (str) – Name of the engine to remove

Returns:

True if engine was removed, False if not found

Return type:

bool

setup_engines_and_tools()

Setup engines and sync their tools, structured output models, and add engine to state.

This validator runs after the model is created and: 1. Finds all engine fields in the state 2. Syncs engine to main engine field and engines dict 3. Syncs tools from engine to state tools field 4. Syncs structured output models 5. Sets up parent-child relationships for nested state schemas

Return type:

Self

sync_engine_fields()

Sync between engine and engines dict for backward compatibility.

This validator ensures that: 1. If ‘engine’ is set, it’s available in engines dict 2. If engines dict has items but no engine, set main engine 3. Both access patterns work seamlessly

Return type:

Self

to_command(goto=None, graph=None)

Convert state to a Command object for LangGraph control flow.

Parameters:

  • goto (str | None) – Optional next node to go to

  • graph (str | None) – Optional graph to target (None for current, PARENT for parent)

Returns:

Command object with state update

Return type:

Any

to_dict()

Convert the state to a clean dictionary.

Returns:

Dictionary representation of the state

Return type:

FieldMapping

to_json()

Convert state to JSON string.

Returns:

JSON string representation of the state

Return type:

str

to_runnable_config(thread_id=None, **kwargs)

Convert state to a RunnableConfig.

Parameters:

  • thread_id (str | None) – Optional thread ID for the configuration

  • **kwargs – Additional configuration parameters

Returns:

RunnableConfig containing state data

Return type:

RunnableConfig

update(other)

Update the state with values from another state or dictionary.

This method performs a simple update without applying reducers.

Parameters:

other (dict[str, Any] | StateSchema) – Dictionary or StateSchema with update values

Returns:

Self for chaining

Return type:

StateSchema

agent_metadata: Dict[str, Any]
engine: TEngine | None
engines: builtins.dict[str, Engine]
execution_path: List[str]
graph_context: Dict[str, Any]
property inputs: Dict[str, Any]

Backward compatibility for inputs.

legacy_inputs: Dict[str, Any]
legacy_meta: Dict[str, Any]
legacy_outputs: Dict[str, Any]
legacy_state: Dict[str, Any]
property llm: Engine | None

Convenience property to access the LLM engine.

property main_engine: Engine | None

Convenience property to access the main engine.

property meta: Dict[str, Any]

Backward compatibility for meta.

model_computed_fields = {}
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property model_extra: dict[str, Any] | None

Get extra fields set during validation.

Returns:

A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'agent_metadata': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Metadata about agents in execution'), 'engine': FieldInfo(annotation=Union[TypeVar, NoneType], required=False, default=None, description='Optional main/primary engine for convenience'), 'engines': FieldInfo(annotation=dict[str, Engine], required=False, default_factory=dict, description='Engine registry for this state - supports easy addition'), 'execution_path': FieldInfo(annotation=List[str], required=False, default_factory=list, description='Path of agents/nodes executed'), 'graph_context': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Graph-level execution context'), 'legacy_inputs': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_meta': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_outputs': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict), 'legacy_state': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict)}
property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,

i.e. that were not filled from defaults.

property outputs: Dict[str, Any]

Backward compatibility for outputs.

property state: Dict[str, Any]

Backward compatibility for state.

Type Definitions

Enums

Type Aliases

# Backward compatibility aliases
AgentGraph = HAPGraph  # Old name for HAPGraph
AgentNode = HAPNode    # Old name for HAPNode

Exceptions

Utilities

Graph Utilities

Context Utilities

Constants

haive.hap.VERSION = "1.0.0"

Current HAP version

haive.hap.PROTOCOL_VERSION = "2.0"

JSON-RPC protocol version for HAP

Complete Example

Here’s a complete example using the full API:

import asyncio
from haive.hap.models import HAPGraph, HAPContext
from haive.hap.server.runtime import HAPRuntime
from haive.hap.hap import HAPContext, SimpleAuthProvider
from haive.agents.simple.agent import SimpleAgent
from haive.core.engine.aug_llm import AugLLMConfig

async def complete_example():
    # Create agents
    agent1 = SimpleAgent(
        name="processor",
        engine=AugLLMConfig(temperature=0.5)
    )

    agent2 = SimpleAgent(
        name="validator",
        engine=AugLLMConfig(temperature=0.3)
    )

    # Build graph
    graph = HAPGraph()
    graph.add_agent_node("process", agent1, ["validate"])
    graph.add_agent_node("validate", agent2)
    graph.entry_node = "process"

    # Create runtime
    runtime = HAPRuntime(graph)

    # Create HAP context for protocol layer
    hap_context = HAPContext(
        request_id="req-123",
        auth_provider=SimpleAuthProvider(
            user="alice",
            scopes=["execute"]
        )
    )

    # Execute with initial context
    initial_data = {
        "input": "Process this data",
        "hap_context": hap_context
    }

    # Run workflow
    result = await runtime.run(initial_data)

    # Access results
    print(f"Execution path: {result.execution_path}")
    print(f"Final output: {result.outputs}")
    print(f"Metadata: {result.agent_metadata}")

    return result

# Run the example
if __name__ == "__main__":
    asyncio.run(complete_example())