API

econagents: A Python library that lets you use LLM agents in economic experiments.

class econagents.Agent(*, url: str, state: GameState, role: Role, prompts_dir: Path, phase_transition_event: str = 'phase-transition', phase_identifier_key: str = 'phase', phase_engine: PhaseEngine | None = None, message_codec: MessageCodec | None = None, state_projector: StateProjectorPort | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, end_game_event: str = 'game-over', logger: Logger | None = None, transport: TransportPort | None = None)

Bases: LoggerMixin

Run one agent against a game server.

property llm_provider

Return the LLM provider used by the role.

register_event_handler(event_type: str, handler: Callable[[Event], Any]) → Agent

Register a handler that runs after state projection.

register_phase_handler(phase: int | str, handler: Callable[[int | str, GameState], Any]) → Agent

Register a handler for a phase.

async start() → None

Connect to the game server and process events until stopped.

async stop() → None

Stop the agent and transport.

async on_event(event: Event) → None

Project an event into state and run the relevant behavior.

async handle_phase_transition(phase: int | str | None) → None

Move to a new phase and execute the appropriate action behavior.

async execute_phase_action(phase: int | str) → None

Execute one action for a phase.

pydantic model econagents.Action

Bases: BaseModel

An agent action before protocol encoding.

Fields:

• payload (dict[str, Any])

• raw (dict[str, Any] | None)

• type (str | None)

field type: str | None = None

field payload: dict[str, Any] [Optional]

field raw: dict[str, Any] | None = None

as_payload() → dict[str, Any]

Return the dict payload that should be encoded for transport.

class econagents.Role(logger: Logger | None = None, persona: Persona | None = None, prompt_renderer: PromptRendererPort | None = None, response_parser: ResponseParserPort | None = None, tools: list[Tool] | None = None)

Bases: ABC, Generic[StateT_contra], LoggerMixin

Base role class with common attributes and phase handling.

This class provides a flexible framework for handling different phases in a game or task workflow. It uses injected prompt rendering and response parsing ports and allows customization for specific phases.

Parameters:

logger (Optional[logging.Logger]) -- External logger to use, defaults to None

role: ClassVar[int]

Unique identifier for this role

name: ClassVar[str]

Human-readable name for this role

llm: LLMProvider

Language model instance for generating responses

task_phases: ClassVar[list[int | str]] = []

List of phases this agent should participate in (empty means all phases)

task_phases_excluded: ClassVar[list[int | str]] = []

Alternative way to specify phases this agent should participate in, listed phases are excluded (empty means nothing excluded)

response_schemas: ClassVar[Dict[int | str, Type[BaseModel]]] = {}

Phase-specific Pydantic schemas used as structured output formats.

default_response_schema: ClassVar[Type[BaseModel] | None] = None

Fallback schema used for phases not listed in response_schemas.

tools: ClassVar[list[Tool]] = []

Read-only tools the LLM may call while handling a phase. Override per role, or pass tools= to the constructor.

max_tool_iterations: ClassVar[int] = 5

Safety cap on tool-call rounds per LLM response.

auto_render_persona: ClassVar[bool] = True

When True and a persona is attached, append a standard markdown block describing the persona to the end of the system prompt. Set to False to take full control via {{ persona }} in your own template.

persona: Persona | None = None

Optional persona injected into the prompt context as persona.

prompt_renderer: PromptRendererPort | None = None

Renderer used by the default prompt path.

response_parser: ResponseParserPort | None = None

Parser used by the default LLM response path.

render_prompt(context: dict, prompt_type: Literal['system', 'user'], phase: int | str, prompts_path: Path) → str

Render a prompt template with the given context.

Template resolution order:

1. Role-specific phase prompt (e.g., "role_name_system_phase_1.jinja2")

2. Role-specific general prompt (e.g., "role_name_system.jinja2")

3. All-role phase prompt (e.g., "all_system_phase_1.jinja2")

4. All-role general prompt (e.g., "all_system.jinja2")

Parameters:

• context (dict) -- Template context variables

• prompt_type -- Type of prompt (system, user)

• phase (int) -- Game phase number

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Rendered prompt

Return type:

str

Raises:

FileNotFoundError -- If no matching prompt template is found

register_system_prompt_handler(phase: int | str, handler: Callable[[StateT_contra], str]) → None

Register a custom system prompt handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (SystemPromptHandler) -- Function that generates system prompts for this phase

register_user_prompt_handler(phase: int | str, handler: Callable[[StateT_contra], str]) → None

Register a custom user prompt handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (UserPromptHandler) -- Function that generates user prompts for this phase

register_response_parser(phase: int | str, parser: Callable[[str | BaseModel, StateT_contra], dict]) → None

Register a custom response parser for a specific phase.

Parameters:

• phase (int) -- Game phase number

• parser (ResponseParser) -- Function that parses LLM responses for this phase

register_response_schema(phase: int | str, schema: Type[BaseModel]) → None

Register a Pydantic response schema for a specific phase.

When a schema is registered, the LLM is asked to emit structured output matching it, and the parsed instance is used as the phase result.

Parameters:

• phase (int) -- Game phase number

• schema (Type[BaseModel]) -- Pydantic model describing the output

get_response_schema(phase: int | str) → Type[BaseModel] | None

Return the schema to use for a given phase, if any.

register_phase_handler(phase: int | str, handler: Callable[[int | str, StateT_contra], Any]) → None

Register a custom phase handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (PhaseHandler) -- Function that handles this phase

get_phase_system_prompt(state: StateT_contra, prompts_path: Path) → str

Get the system prompt for the current phase.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using templates.

Parameters:

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

System prompt string

Return type:

str

get_phase_user_prompt(state: StateT_contra, prompts_path: Path) → str

Get the user prompt for the current phase.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using templates.

Parameters:

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

User prompt string

Return type:

str

parse_phase_llm_response(response: str | BaseModel, state: StateT_contra) → dict

Parse the LLM response for the current phase.

Resolution order:

1. A phase-specific parser registered via register_response_parser.

2. If the provider returned a validated Pydantic instance, its model_dump().

3. If a response schema is registered for this phase (or a default schema is set), validate the raw string against it.

4. Fall back to json.loads on the raw string.

Parameters:

• response -- Either a raw LLM response string or a Pydantic instance produced by a structured-output-capable provider.

• state -- Current game state.

Returns:

Parsed response as a dictionary.

Return type:

dict

async handle_phase(phase: int | str, state: StateT_contra, prompts_path: Path) → dict | None

Handle the current phase of the task or game.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using the LLM.

By default, the agent acts in all phases unless: 1. task_phases is non-empty and the phase is not in task_phases, or 2. phase is explicitly listed in task_phases_excluded

Parameters:

• phase (int) -- Game phase number

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Phase result dictionary or None if phase is not handled

Return type:

Optional[dict]

async handle_phase_with_llm(phase: int | str, state: StateT_contra, prompts_path: Path) → dict | None

Handle the phase using the LLM.

This is the default implementation that uses the LLM to handle the phase by generating prompts, sending them to the LLM, and parsing the response.

Parameters:

• phase (int) -- Game phase number

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Phase result dictionary or None if phase is not handled

Return type:

Optional[dict]

pydantic model econagents.AgentContext

Bases: BaseModel

Stable identity for one simulated player in an experiment.

Fields:

• agent_id (int | str)

• game_id (int)

• role_id (int | None)

field game_id: int [Required]

field agent_id: int | str [Required]

field role_id: int | None = None

class econagents.YamlExperimentLoader(config_path: Path)

Bases: object

Load and run an experiment specification from YAML.

load_config() → ExperimentSpec

Load the experiment configuration from the YAML file.

async run_experiment(login_payloads: List[Dict[str, Any]], game_id: int) → None

Run the experiment from this configuration.

Parameters:

login_payloads -- A list of dictionaries containing login information for each agent

pydantic model econagents.Event

Bases: BaseModel

An event observed by an agent after protocol decoding.

Config:

• arbitrary_types_allowed: bool = True

Fields:

• data (dict[str, Any])

• raw (Any | None)

• source (str | None)

• type (str)

field type: str [Required]

field data: dict[str, Any] [Optional]

field source: str | None = None

field raw: Any | None = None

econagents.EventField(default: Any = Ellipsis, *, default_factory: Callable[[], Any] | None = None, event_key: str | None = None, exclude_from_mapping: bool = False, events: list[str] | None = None, exclude_events: list[str] | None = None, **kwargs: Any) → Any

Create a field with event mapping metadata.

Parameters:

• default (Any) -- Default value for the field

• default_factory (Callable[[], Any]) -- Factory function to generate default value

• event_key (Optional[str]) -- The key in event data that maps to this field

• exclude_from_mapping (bool) -- Whether to exclude this field from event mapping

• events (Optional[list[str]]) -- Optional list of events where this mapping should be applied

• exclude_events (Optional[list[str]]) -- Optional list of events where this mapping should not be applied

• **kwargs -- Additional arguments to pass to Pydantic's Field

Returns:

A Pydantic FieldInfo object with event mapping metadata

Return type:

FieldInfo

class econagents.GameRunner(config: GameRunnerConfig, agents: list[Agent])

Bases: object

get_agent_logger(agent_id: int, game_id: int) → Logger

Configure and return a logger for an agent.

Parameters:

• agent_id (int) -- Agent identifier

• game_id (int) -- Game identifier

Returns:

Configured logger instance

Return type:

logging.Logger

get_game_logger(game_id: int) → Logger

Configure and return a logger for a game.

Parameters:

game_id (int) -- Game identifier

Returns:

Configured logger instance

Return type:

logging.Logger

cleanup_logging() → None

Clean up logging resources, stopping all queue listeners. Should be called when shutting down the game runner.

async spawn_agent(agent: Agent, agent_id: int) → None

Spawn an agent and connect it to the game.

Parameters:

• agent -- Agent to spawn

• agent_id (int) -- Agent identifier

async run_game() → None

Run a game using provided game data.

pydantic model econagents.GameState

Bases: BaseModel

Game state for a given game

Fields:

• meta (econagents.domain.state.game.MetaInformation)

• private_information (econagents.domain.state.game.PrivateInformation)

• public_information (econagents.domain.state.game.PublicInformation)

field meta: MetaInformation [Optional]

Meta information for the game

field private_information: PrivateInformation [Optional]

Private information for each agent in the game

field public_information: PublicInformation [Optional]

Public information for the game

update(event: Message) → None

Generic state update method that handles both property mappings and custom event handlers.

Parameters:

event (Message) -- The event message containing event_type and data

This method will: 1. Check for custom event handlers first 2. Fall back to property mappings if no custom handler exists 3. Update state based on property mappings, considering phase restrictions

get_custom_handlers() → dict[str, Callable[[str, dict[str, Any]], None]]

Override this method to provide custom event handlers.

Returns:

A mapping of event types to handler functions.

Return type:

dict[str, EventHandler]

reset() → None

Resets meta, private_information, and public_information to their initial state by re-initializing them using their default factories. This effectively removes any dynamically added attributes.

pydantic model econagents.HybridGameRunnerConfig

Bases: GameRunnerConfig

Configuration class for TurnBasedGameRunner.

Fields:

• continuous_phases (list[int | str])

• max_action_delay (int)

• min_action_delay (int)

field continuous_phases: list[int | str] [Optional]

field min_action_delay: int = 5

field max_action_delay: int = 10

class econagents.IbexMessageCodec

Bases: object

Translate IBEX WebSocket envelopes to and from domain messages.

Inbound messages use the IBEX shape {"meta": {"type": ...}, "payload": {...}}.

decode_event(raw_message: str) → Event

Decode an inbound wire message into a domain event.

encode_action(action: dict[str, Any] | Action) → str

Encode an action as a JSON string.

Dict actions are treated as already-shaped outbound payloads.

encode_join(payload: dict[str, Any]) → str

Encode an IBEX join envelope.

encode_ready() → str

Encode an IBEX ready envelope.

class econagents.JoinPayloadAuth

Bases: AuthenticationMechanism

Default authentication mechanism.

Sends a join envelope as the first message on the connection:

{"meta": {"type": "join"}, "payload": {<kwargs>}}

The keyword arguments passed via auth_mechanism_kwargs become the payload (typically {"recovery": "<code>"}). If the kwargs already contain a meta key they are treated as a fully-formed envelope and sent as-is, so callers may still pass an explicit envelope when needed.

async authenticate(transport: WebSocketTransport, **kwargs) → bool

Send the join envelope as a JSON message.

pydantic model econagents.MetaInformation

Bases: BaseModel

Meta information for the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

Fields:

• game_id (int)

• phase (int | str)

• player_name (str | None)

• player_number (int | None)

• players (list[dict[str, Any]])

field game_id: int = 0

ID of the game

field player_name: str | None = None

Name of the player

field player_number: int | None = None

Number of the player

field players: list[dict[str, Any]] [Optional]

List of players in the game

field phase: int | str = 0

Current phase of the game

class econagents.PhaseEngine(continuous_phases: set[int | str] | None = None, min_action_delay: int | None = None, max_action_delay: int | None = None, random_int: Callable[[int, int], int] | None = None)

Bases: object

Decide whether phases are continuous and when repeated actions occur.

is_continuous(phase: int | str) → bool

Return whether a phase should run repeated actions.

next_action_delay() → int

Return the delay before the next action in a continuous phase.

pydantic model econagents.PrivateInformation

Bases: BaseModel

Private information for each agent in the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

pydantic model econagents.PublicInformation

Bases: BaseModel

Public information for the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

class econagents.SimpleLoginPayloadAuth

Bases: AuthenticationMechanism

Authentication mechanism that sends a login payload as the first message.

async authenticate(transport: WebSocketTransport, **kwargs) → bool

Send the login payload as a JSON message.

pydantic model econagents.TurnBasedGameRunnerConfig

Bases: GameRunnerConfig

Configuration class for TurnBasedGameRunner.

Fields:

class econagents.WebSocketTransport(url: str, logger: Logger | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, on_message_callback: Callable[[str], Any] | None = None)

Bases: LoggerMixin

Responsible for connecting to a WebSocket, sending/receiving messages, and reporting received messages to a callback function.

async start_listening()

Begin receiving messages in a loop.

async send(message: str)

Send a raw string message to the WebSocket.

async stop()

Gracefully close the WebSocket connection.

econagents.build_message(type: str, payload: dict[str, Any] | None = None, component: str | dict[str, Any] | None = None) → dict[str, Any]

Build an IBEX message envelope.

econagents.create_game_state(state_type: Type[StateT], game_id: int, **kwargs) → StateT

Create a game state instance for a game.

econagents.join_message(**payload: Any) → dict[str, Any]

Build the IBEX join envelope.

econagents.ready_message() → dict[str, Any]

Build the IBEX ready envelope.

Domain

Domain-level types used by the econagents runtime.

pydantic model econagents.domain.Action

Bases: BaseModel

An agent action before protocol encoding.

Fields:

• payload (dict[str, Any])

• raw (dict[str, Any] | None)

• type (str | None)

field type: str | None = None

field payload: dict[str, Any] [Optional]

field raw: dict[str, Any] | None = None

as_payload() → dict[str, Any]

Return the dict payload that should be encoded for transport.

pydantic model econagents.domain.AgentContext

Bases: BaseModel

Stable identity for one simulated player in an experiment.

Fields:

• agent_id (int | str)

• game_id (int)

• role_id (int | None)

field game_id: int [Required]

field agent_id: int | str [Required]

field role_id: int | None = None

class econagents.domain.Role(logger: Logger | None = None, persona: Persona | None = None, prompt_renderer: PromptRendererPort | None = None, response_parser: ResponseParserPort | None = None, tools: list[Tool] | None = None)

Bases: ABC, Generic[StateT_contra], LoggerMixin

Base role class with common attributes and phase handling.

This class provides a flexible framework for handling different phases in a game or task workflow. It uses injected prompt rendering and response parsing ports and allows customization for specific phases.

Parameters:

logger (Optional[logging.Logger]) -- External logger to use, defaults to None

role: ClassVar[int]

Unique identifier for this role

name: ClassVar[str]

Human-readable name for this role

llm: LLMProvider

Language model instance for generating responses

task_phases: ClassVar[list[int | str]] = []

List of phases this agent should participate in (empty means all phases)

task_phases_excluded: ClassVar[list[int | str]] = []

Alternative way to specify phases this agent should participate in, listed phases are excluded (empty means nothing excluded)

response_schemas: ClassVar[Dict[int | str, Type[BaseModel]]] = {}

Phase-specific Pydantic schemas used as structured output formats.

default_response_schema: ClassVar[Type[BaseModel] | None] = None

Fallback schema used for phases not listed in response_schemas.

tools: ClassVar[list[Tool]] = []

Read-only tools the LLM may call while handling a phase. Override per role, or pass tools= to the constructor.

max_tool_iterations: ClassVar[int] = 5

Safety cap on tool-call rounds per LLM response.

auto_render_persona: ClassVar[bool] = True

When True and a persona is attached, append a standard markdown block describing the persona to the end of the system prompt. Set to False to take full control via {{ persona }} in your own template.

persona: Persona | None = None

Optional persona injected into the prompt context as persona.

prompt_renderer: PromptRendererPort | None = None

Renderer used by the default prompt path.

response_parser: ResponseParserPort | None = None

Parser used by the default LLM response path.

render_prompt(context: dict, prompt_type: Literal['system', 'user'], phase: int | str, prompts_path: Path) → str

Render a prompt template with the given context.

Template resolution order:

1. Role-specific phase prompt (e.g., "role_name_system_phase_1.jinja2")

2. Role-specific general prompt (e.g., "role_name_system.jinja2")

3. All-role phase prompt (e.g., "all_system_phase_1.jinja2")

4. All-role general prompt (e.g., "all_system.jinja2")

Parameters:

• context (dict) -- Template context variables

• prompt_type -- Type of prompt (system, user)

• phase (int) -- Game phase number

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Rendered prompt

Return type:

str

Raises:

FileNotFoundError -- If no matching prompt template is found

register_system_prompt_handler(phase: int | str, handler: Callable[[StateT_contra], str]) → None

Register a custom system prompt handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (SystemPromptHandler) -- Function that generates system prompts for this phase

register_user_prompt_handler(phase: int | str, handler: Callable[[StateT_contra], str]) → None

Register a custom user prompt handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (UserPromptHandler) -- Function that generates user prompts for this phase

register_response_parser(phase: int | str, parser: Callable[[str | BaseModel, StateT_contra], dict]) → None

Register a custom response parser for a specific phase.

Parameters:

• phase (int) -- Game phase number

• parser (ResponseParser) -- Function that parses LLM responses for this phase

register_response_schema(phase: int | str, schema: Type[BaseModel]) → None

Register a Pydantic response schema for a specific phase.

When a schema is registered, the LLM is asked to emit structured output matching it, and the parsed instance is used as the phase result.

Parameters:

• phase (int) -- Game phase number

• schema (Type[BaseModel]) -- Pydantic model describing the output

get_response_schema(phase: int | str) → Type[BaseModel] | None

Return the schema to use for a given phase, if any.

register_phase_handler(phase: int | str, handler: Callable[[int | str, StateT_contra], Any]) → None

Register a custom phase handler for a specific phase.

Parameters:

• phase (int) -- Game phase number

• handler (PhaseHandler) -- Function that handles this phase

get_phase_system_prompt(state: StateT_contra, prompts_path: Path) → str

Get the system prompt for the current phase.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using templates.

Parameters:

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

System prompt string

Return type:

str

get_phase_user_prompt(state: StateT_contra, prompts_path: Path) → str

Get the user prompt for the current phase.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using templates.

Parameters:

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

User prompt string

Return type:

str

parse_phase_llm_response(response: str | BaseModel, state: StateT_contra) → dict

Parse the LLM response for the current phase.

Resolution order:

1. A phase-specific parser registered via register_response_parser.

2. If the provider returned a validated Pydantic instance, its model_dump().

3. If a response schema is registered for this phase (or a default schema is set), validate the raw string against it.

4. Fall back to json.loads on the raw string.

Parameters:

• response -- Either a raw LLM response string or a Pydantic instance produced by a structured-output-capable provider.

• state -- Current game state.

Returns:

Parsed response as a dictionary.

Return type:

dict

async handle_phase(phase: int | str, state: StateT_contra, prompts_path: Path) → dict | None

Handle the current phase of the task or game.

This method will use a phase-specific handler if registered, otherwise it falls back to the default implementation using the LLM.

By default, the agent acts in all phases unless: 1. task_phases is non-empty and the phase is not in task_phases, or 2. phase is explicitly listed in task_phases_excluded

Parameters:

• phase (int) -- Game phase number

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Phase result dictionary or None if phase is not handled

Return type:

Optional[dict]

async handle_phase_with_llm(phase: int | str, state: StateT_contra, prompts_path: Path) → dict | None

Handle the phase using the LLM.

This is the default implementation that uses the LLM to handle the phase by generating prompts, sending them to the LLM, and parsing the response.

Parameters:

• phase (int) -- Game phase number

• state (StateT_contra) -- Current game state

• prompts_path (Path) -- Path to prompt templates directory

Returns:

Phase result dictionary or None if phase is not handled

Return type:

Optional[dict]

pydantic model econagents.domain.Event

Bases: BaseModel

An event observed by an agent after protocol decoding.

Config:

• arbitrary_types_allowed: bool = True

Fields:

• data (dict[str, Any])

• raw (Any | None)

• source (str | None)

• type (str)

field type: str [Required]

field data: dict[str, Any] [Optional]

field source: str | None = None

field raw: Any | None = None

econagents.domain.EventField(default: Any = Ellipsis, *, default_factory: Callable[[], Any] | None = None, event_key: str | None = None, exclude_from_mapping: bool = False, events: list[str] | None = None, exclude_events: list[str] | None = None, **kwargs: Any) → Any

Create a field with event mapping metadata.

Parameters:

• default (Any) -- Default value for the field

• default_factory (Callable[[], Any]) -- Factory function to generate default value

• event_key (Optional[str]) -- The key in event data that maps to this field

• exclude_from_mapping (bool) -- Whether to exclude this field from event mapping

• events (Optional[list[str]]) -- Optional list of events where this mapping should be applied

• exclude_events (Optional[list[str]]) -- Optional list of events where this mapping should not be applied

• **kwargs -- Additional arguments to pass to Pydantic's Field

Returns:

A Pydantic FieldInfo object with event mapping metadata

Return type:

FieldInfo

pydantic model econagents.domain.GameState

Bases: BaseModel

Game state for a given game

Fields:

• meta (econagents.domain.state.game.MetaInformation)

• private_information (econagents.domain.state.game.PrivateInformation)

• public_information (econagents.domain.state.game.PublicInformation)

field meta: MetaInformation [Optional]

Meta information for the game

field private_information: PrivateInformation [Optional]

Private information for each agent in the game

field public_information: PublicInformation [Optional]

Public information for the game

update(event: Message) → None

Generic state update method that handles both property mappings and custom event handlers.

Parameters:

event (Message) -- The event message containing event_type and data

This method will: 1. Check for custom event handlers first 2. Fall back to property mappings if no custom handler exists 3. Update state based on property mappings, considering phase restrictions

get_custom_handlers() → dict[str, Callable[[str, dict[str, Any]], None]]

Override this method to provide custom event handlers.

Returns:

A mapping of event types to handler functions.

Return type:

dict[str, EventHandler]

reset() → None

Resets meta, private_information, and public_information to their initial state by re-initializing them using their default factories. This effectively removes any dynamically added attributes.

class econagents.domain.GameStateProtocol(*args, **kwargs)

Bases: Protocol

meta: MetaInformation

private_information: PrivateInformation

public_information: PublicInformation

model_dump() → dict[str, Any]

model_dump_json() → str

pydantic model econagents.domain.Message

Bases: BaseModel

A message from the server to the agent.

Fields:

• data (dict[str, Any])

• event_type (str)

• message_type (str)

field message_type: str [Required]

Type of message

field event_type: str [Required]

Type of event

field data: dict[str, Any] [Required]

Data associated with the message

classmethod from_event(event: Event) → Message

Create a transport message from a domain event.

to_event() → Event

Convert this transport message to a domain event.

pydantic model econagents.domain.MetaInformation

Bases: BaseModel

Meta information for the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

Fields:

• game_id (int)

• phase (int | str)

• player_name (str | None)

• player_number (int | None)

• players (list[dict[str, Any]])

field game_id: int = 0

ID of the game

field player_name: str | None = None

Name of the player

field player_number: int | None = None

Number of the player

field players: list[dict[str, Any]] [Optional]

List of players in the game

field phase: int | str = 0

Current phase of the game

pydantic model econagents.domain.PrivateInformation

Bases: BaseModel

Private information for each agent in the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

pydantic model econagents.domain.PropertyMapping

Bases: BaseModel

Mapping between event data and state properties

Parameters:

• event_key -- Key in the event data

• state_key -- Key in the state object

• state_type -- Whether to update private or public information ("private" or "public")

• phases -- Optional list of phases where this mapping should be applied. If None, applies to all phases.

• exclude_phases -- Optional list of phases where this mapping should not be applied. Cannot be used together with phases.

Fields:

• event_key (str)

• events (list[str] | None)

• exclude_events (list[str] | None)

• state_key (str)

• state_type (str)

field event_key: str [Required]

field state_key: str [Required]

field state_type: str = 'private'

field events: list[str] | None = None

field exclude_events: list[str] | None = None

model_post_init(_PropertyMapping__context: Any) → None

Validate that events and exclude_events are not both specified

should_apply_in_event(current_event: str) → bool

Determine if this mapping should be applied in the current event

pydantic model econagents.domain.PublicInformation

Bases: BaseModel

Public information for the game

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = False

Runtime

Runtime services that coordinate domain objects and ports.

class econagents.runtime.Agent(*, url: str, state: GameState, role: Role, prompts_dir: Path, phase_transition_event: str = 'phase-transition', phase_identifier_key: str = 'phase', phase_engine: PhaseEngine | None = None, message_codec: MessageCodec | None = None, state_projector: StateProjectorPort | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, end_game_event: str = 'game-over', logger: Logger | None = None, transport: TransportPort | None = None)

Bases: LoggerMixin

Run one agent against a game server.

property llm_provider

Return the LLM provider used by the role.

register_event_handler(event_type: str, handler: Callable[[Event], Any]) → Agent

Register a handler that runs after state projection.

register_phase_handler(phase: int | str, handler: Callable[[int | str, GameState], Any]) → Agent

Register a handler for a phase.

async start() → None

Connect to the game server and process events until stopped.

async stop() → None

Stop the agent and transport.

async on_event(event: Event) → None

Project an event into state and run the relevant behavior.

async handle_phase_transition(phase: int | str | None) → None

Move to a new phase and execute the appropriate action behavior.

async execute_phase_action(phase: int | str) → None

Execute one action for a phase.

class econagents.runtime.GameRunner(config: GameRunnerConfig, agents: list[Agent])

Bases: object

get_agent_logger(agent_id: int, game_id: int) → Logger

Configure and return a logger for an agent.

Parameters:

• agent_id (int) -- Agent identifier

• game_id (int) -- Game identifier

Returns:

Configured logger instance

Return type:

logging.Logger

get_game_logger(game_id: int) → Logger

Configure and return a logger for a game.

Parameters:

game_id (int) -- Game identifier

Returns:

Configured logger instance

Return type:

logging.Logger

cleanup_logging() → None

Clean up logging resources, stopping all queue listeners. Should be called when shutting down the game runner.

async spawn_agent(agent: Agent, agent_id: int) → None

Spawn an agent and connect it to the game.

Parameters:

• agent -- Agent to spawn

• agent_id (int) -- Agent identifier

async run_game() → None

Run a game using provided game data.

pydantic model econagents.runtime.GameRunnerConfig

Bases: BaseModel

Configuration class for GameRunner.

Fields:

• auth_mechanism (econagents.adapters.transport.websocket.AuthenticationMechanism | None)

• end_game_event (str)

• game_id (int)

• hostname (str)

• log_level (int)

• logs_dir (pathlib.Path)

• max_game_duration (int)

• observability_provider (Literal['langsmith', 'langfuse'] | None)

• path (str)

• phase_identifier_key (str)

• phase_transition_event (str)

• port (int)

• prompts_dir (pathlib.Path)

• protocol (str)

field protocol: str = 'ws'

Protocol to use for the server

field hostname: str [Required]

Hostname of the server

field path: str [Required]

Path to the server

field port: int [Required]

field game_id: int [Required]

ID of the game

field logs_dir: Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/econagents/checkouts/stable/docs/source/logs')

Directory to store logs

field log_level: int = 20

Level of logging to use

field prompts_dir: Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/econagents/checkouts/stable/docs/source/prompts')

field auth_mechanism: AuthenticationMechanism | None = <econagents.adapters.transport.websocket.JoinPayloadAuth object>

Authentication mechanism to use. Defaults to the join handshake.

field phase_transition_event: str = 'phase-transition'

Event to use for phase transitions

field phase_identifier_key: str = 'phase'

Key in data to use for phase identification

field observability_provider: Literal['langsmith', 'langfuse'] | None = None

Name of the observability provider to use. Options: 'langsmith' or 'langfuse'

field max_game_duration: int = 600

Maximum game duration in seconds. Default is 600 (10 minutes). Set to 0 or a negative value to disable the timeout.

field end_game_event: str = 'game-over'

Event type that signals the end of the game and should stop the agent.

server_url() → str

Build the WebSocket URL for agents.

pydantic model econagents.runtime.HybridGameRunnerConfig

Bases: GameRunnerConfig

Configuration class for TurnBasedGameRunner.

Fields:

• auth_mechanism (Optional[AuthenticationMechanism])

• continuous_phases (list[int | str])

• end_game_event (str)

• game_id (int)

• hostname (str)

• log_level (int)

• logs_dir (Path)

• max_action_delay (int)

• max_game_duration (int)

• min_action_delay (int)

• observability_provider (Optional[Literal['langsmith', 'langfuse']])

• path (str)

• phase_identifier_key (str)

• phase_transition_event (str)

• port (int)

• prompts_dir (Path)

• protocol (str)

field continuous_phases: list[int | str] [Optional]

field min_action_delay: int = 5

field max_action_delay: int = 10

field protocol: str = 'ws'

Protocol to use for the server

field hostname: str [Required]

Hostname of the server

field path: str [Required]

Path to the server

field port: int [Required]

field game_id: int [Required]

ID of the game

field logs_dir: Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/econagents/checkouts/stable/docs/source/logs')

Directory to store logs

field log_level: int = 20

Level of logging to use

field prompts_dir: Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/econagents/checkouts/stable/docs/source/prompts')

field auth_mechanism: AuthenticationMechanism | None = <econagents.adapters.transport.websocket.JoinPayloadAuth object>

Authentication mechanism to use. Defaults to the join handshake.

field phase_transition_event: str = 'phase-transition'

Event to use for phase transitions

field phase_identifier_key: str = 'phase'

Key in data to use for phase identification

field observability_provider: Literal['langsmith', 'langfuse'] | None = None

Name of the observability provider to use. Options: 'langsmith' or 'langfuse'

field max_game_duration: int = 600

Maximum game duration in seconds. Default is 600 (10 minutes). Set to 0 or a negative value to disable the timeout.

field end_game_event: str = 'game-over'

Event type that signals the end of the game and should stop the agent.

class econagents.runtime.PhaseEngine(continuous_phases: set[int | str] | None = None, min_action_delay: int | None = None, max_action_delay: int | None = None, random_int: Callable[[int, int], int] | None = None)

Bases: object

Decide whether phases are continuous and when repeated actions occur.

is_continuous(phase: int | str) → bool

Return whether a phase should run repeated actions.

next_action_delay() → int

Return the delay before the next action in a continuous phase.

pydantic model econagents.runtime.TurnBasedGameRunnerConfig

Bases: GameRunnerConfig

Configuration class for TurnBasedGameRunner.

Fields:

econagents.runtime.create_game_state(state_type: Type[StateT], game_id: int, **kwargs) → StateT

Create a game state instance for a game.

Ports

Interfaces that separate domain and runtime code from external systems.

class econagents.ports.MessageCodec(*args, **kwargs)

Bases: Protocol

Translate between external wire messages and internal events/actions.

decode_event(raw_message: str) → Event

Decode a raw inbound message into a domain event.

encode_action(action: dict[str, Any] | Action) → str

Encode an outbound action for the transport.

encode_join(payload: dict[str, Any]) → str

Encode an authentication/join message.

encode_ready() → str

Encode the standard ready message.

exception econagents.ports.MessageDecodeError

Bases: ValueError

Raised when a raw transport message cannot be decoded.

class econagents.ports.LLMProvider(*args, **kwargs)

Bases: Protocol

Interface for model providers used by roles.

async get_response(messages: list[dict[str, Any]], tracing_extra: dict[str, Any], response_schema: Type[BaseModel] | None = None, tools: list[ToolSpec] | None = None, tool_executor: ToolExecutor | None = None, max_tool_iterations: int = 5) → str | BaseModel

Return a raw model response or a validated structured response.

When tools and tool_executor are provided, the adapter runs the provider-native tool-calling loop: it advertises the tools, executes any requested calls via tool_executor, feeds the results back, and repeats up to max_tool_iterations times until the model returns a final answer. The return type is unchanged whether or not tools are used.

build_messages(system_prompt: str, user_prompt: str) → list[dict[str, Any]]

Build provider-specific chat messages.

class econagents.ports.PromptRendererPort(*args, **kwargs)

Bases: Protocol

Render phase prompts from state/context without owning agent policy.

render(context: dict[str, Any], prompt_type: Literal['system', 'user'], phase: int | str, prompts_path: Path, role_names: list[str], resolver: Callable[[Literal['system', 'user'], int | str, str, Path], Path | None] | None = None, logger: Any | None = None) → str

Render a prompt from the configured prompt source.

class econagents.ports.ResponseParserPort(*args, **kwargs)

Bases: Protocol

Validate and convert provider responses into action dictionaries.

parse(response: str | BaseModel, state: GameStateProtocol, phase: int | str, response_schema: Type[BaseModel] | None = None, logger: Any | None = None) → dict[str, Any]

Parse the provider response for a phase.

class econagents.ports.StateProjectorPort(*args, **kwargs)

Bases: Protocol

Apply domain events to an agent-local game state.

apply(state: GameState, event: Event) → None

Update state in response to an event.

class econagents.ports.Tool(*args, **kwargs)

Bases: Protocol

Interface for a read-only tool an agent can call.

name: str

description: str

spec() → ToolSpec

Return the provider-agnostic spec advertised to the model.

async run(arguments: dict[str, Any], ctx: ToolContext) → Any

Execute the tool and return a JSON-serializable result.

class econagents.ports.ToolCall(id: str, name: str, arguments: dict[str, Any])

Bases: object

A single tool invocation requested by the model.

id: str

Provider call id, echoed back when returning the result.

name: str

arguments: dict[str, Any]

class econagents.ports.ToolContext(state: GameStateProtocol, phase: int | str, logger: Any | None = None)

Bases: object

Read-only context handed to a tool when it runs.

state: GameStateProtocol

phase: int | str

logger: Any | None = None

class econagents.ports.ToolSpec(name: str, description: str, parameters: dict[str, Any])

Bases: object

Provider-agnostic description of a callable tool.

name: str

description: str

parameters: dict[str, Any]

JSON Schema for the tool's arguments.

class econagents.ports.TransportPort(*args, **kwargs)

Bases: Protocol

Minimal async transport interface used by agents.

async start_listening() → None

Start receiving messages.

async send(message: str) → None

Send a raw outbound message.

async stop() → None

Stop the transport.

Protocol Adapters

Protocol adapters.

class econagents.adapters.protocol.FlatMessageCodec

Bases: object

Translate flat JSON messages to and from domain messages.

decode_event(raw_message: str) → Event

Decode a flat inbound message into a domain event.

encode_action(action: dict[str, Any] | Action) → str

Encode an action as a flat JSON string.

class econagents.adapters.protocol.IbexMessageCodec

Bases: object

Translate IBEX WebSocket envelopes to and from domain messages.

Inbound messages use the IBEX shape {"meta": {"type": ...}, "payload": {...}}.

decode_event(raw_message: str) → Event

Decode an inbound wire message into a domain event.

encode_action(action: dict[str, Any] | Action) → str

Encode an action as a JSON string.

Dict actions are treated as already-shaped outbound payloads.

encode_join(payload: dict[str, Any]) → str

Encode an IBEX join envelope.

encode_ready() → str

Encode an IBEX ready envelope.

econagents.adapters.protocol.build_message(type: str, payload: dict[str, Any] | None = None, component: str | dict[str, Any] | None = None) → dict[str, Any]

Build an IBEX message envelope.

econagents.adapters.protocol.join_message(**payload: Any) → dict[str, Any]

Build the IBEX join envelope.

econagents.adapters.protocol.ready_message() → dict[str, Any]

Build the IBEX ready envelope.

Transport Adapters

Transport adapters.

class econagents.adapters.transport.AuthenticationMechanism

Bases: ABC

Abstract base class for authentication mechanisms.

abstract async authenticate(transport: WebSocketTransport, **kwargs) → bool

Authenticate the transport.

class econagents.adapters.transport.JoinPayloadAuth

Bases: AuthenticationMechanism

Default authentication mechanism.

Sends a join envelope as the first message on the connection:

{"meta": {"type": "join"}, "payload": {<kwargs>}}

The keyword arguments passed via auth_mechanism_kwargs become the payload (typically {"recovery": "<code>"}). If the kwargs already contain a meta key they are treated as a fully-formed envelope and sent as-is, so callers may still pass an explicit envelope when needed.

async authenticate(transport: WebSocketTransport, **kwargs) → bool

Send the join envelope as a JSON message.

class econagents.adapters.transport.SimpleLoginPayloadAuth

Bases: AuthenticationMechanism

Authentication mechanism that sends a login payload as the first message.

async authenticate(transport: WebSocketTransport, **kwargs) → bool

Send the login payload as a JSON message.

class econagents.adapters.transport.WebSocketTransport(url: str, logger: Logger | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, on_message_callback: Callable[[str], Any] | None = None)

Bases: LoggerMixin

Responsible for connecting to a WebSocket, sending/receiving messages, and reporting received messages to a callback function.

async start_listening()

Begin receiving messages in a loop.

async send(message: str)

Send a raw string message to the WebSocket.

async stop()

Gracefully close the WebSocket connection.

Config Adapters

Configuration adapters.

pydantic model econagents.adapters.config.AgentSpec

Bases: BaseModel

Configuration for one agent in an experiment.

Optionally attach a persona via persona_id, which references a persona declared in the experiment's top-level personas list.

For file-based or bundled-by-id persona resolution, use the code-driven entry point with econagents.personas.load_persona().

Fields:

• id (int)

• persona_id (str | None)

• role_id (int)

field id: int [Required]

field role_id: int [Required]

field persona_id: str | None = None

pydantic model econagents.adapters.config.RoleSpec

Bases: BaseModel

Configuration for a role.

Fields:

• llm_params (Dict[str, Any])

• llm_type (str)

• name (str)

• prompts (List[Dict[str, str]])

• role_id (int)

• task_phases (List[int | str])

• task_phases_excluded (List[int | str])

field role_id: int [Required]

field name: str [Required]

field llm_type: str = 'ChatOpenAI'

field llm_params: Dict[str, Any] [Optional]

field prompts: List[Dict[str, str]] [Optional]

field task_phases: List[int | str] [Optional]

field task_phases_excluded: List[int | str] [Optional]

create_role(persona: Persona | None = None) → Role

Create a Role instance from this configuration.

pydantic model econagents.adapters.config.EventHandlerSpec

Bases: BaseModel

Configuration for an event handler.

Fields:

• custom_code (str | None)

• custom_function (str | None)

• custom_module (str | None)

• event (str)

field event: str [Required]

field custom_code: str | None = None

field custom_module: str | None = None

field custom_function: str | None = None

pydantic model econagents.adapters.config.ExperimentSpec

Bases: BaseModel

Configuration for an entire experiment.

Fields:

• agents (List[econagents.adapters.config.yaml.AgentSpec])

• description (str)

• name (str)

• personas (List[econagents.personas.Persona])

• prompt_partials (List[Dict[str, str]])

• roles (List[econagents.adapters.config.yaml.RoleSpec])

• runner (econagents.adapters.config.yaml.RunnerSpec)

• runtime (econagents.adapters.config.yaml.RuntimeSpec)

• state (econagents.adapters.config.yaml.StateSpec)

Validators:

• _check_personas_and_references » all fields

field name: str [Required]

Validated by:

• _check_personas_and_references

field description: str = ''

Validated by:

• _check_personas_and_references

field prompt_partials: List[Dict[str, str]] [Optional]

Validated by:

• _check_personas_and_references

field roles: List[RoleSpec] [Optional]

Validated by:

• _check_personas_and_references

field personas: List[Persona] [Optional]

Validated by:

• _check_personas_and_references

field agents: List[AgentSpec] [Optional]

Validated by:

• _check_personas_and_references

field state: StateSpec [Required]

Validated by:

• _check_personas_and_references

field runtime: RuntimeSpec [Required]

Validated by:

• _check_personas_and_references

field runner: RunnerSpec [Required]

Validated by:

• _check_personas_and_references

async run_experiment(login_payloads: List[Dict[str, Any]], game_id: int) → None

Run the experiment from this configuration.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Parameters:

• self -- The BaseModel instance.

• context -- The context.

pydantic model econagents.adapters.config.RunnerSpec

Bases: BaseModel

Configuration for a game runner.

Fields:

• continuous_phases (List[int | str])

• game_id (int)

• hostname (str)

• log_level (str)

• logs_dir (str)

• max_action_delay (int)

• min_action_delay (int)

• observability_provider (Literal['langsmith', 'langfuse'] | None)

• path (str)

• phase_identifier_key (str)

• phase_transition_event (str)

• port (int)

• prompts_dir (str)

• protocol (str)

• type (str)

field type: str = 'GameRunner'

field protocol: str = 'ws'

field hostname: str [Required]

field path: str = 'wss'

field port: int [Required]

field game_id: int [Required]

field logs_dir: str = 'logs'

field log_level: str = 'INFO'

field prompts_dir: str = 'prompts'

field phase_transition_event: str = 'phase-transition'

field phase_identifier_key: str = 'phase'

field observability_provider: Literal['langsmith', 'langfuse'] | None = None

field continuous_phases: List[int | str] [Optional]

field min_action_delay: int = 5

field max_action_delay: int = 10

create_runner_config() → GameRunnerConfig

Create a GameRunnerConfig instance from this configuration.

pydantic model econagents.adapters.config.RuntimeSpec

Bases: BaseModel

Configuration for agents.

Fields:

• event_handlers (List[econagents.adapters.config.yaml.EventHandlerSpec])

• mode (Literal['turn_based', 'hybrid'])

field mode: Literal['turn_based', 'hybrid'] = 'turn_based'

field event_handlers: List[EventHandlerSpec] [Optional]

create_agent(game_id: int, state: GameState, role: Role, auth_kwargs: Dict[str, Any], runner_config: GameRunnerConfig) → Agent

Create a configured agent.

pydantic model econagents.adapters.config.StateSpec

Bases: BaseModel

Configuration for a game state.

Fields:

• meta_information (List[econagents.adapters.config.yaml.StateFieldSpec])

• private_information (List[econagents.adapters.config.yaml.StateFieldSpec])

• public_information (List[econagents.adapters.config.yaml.StateFieldSpec])

field meta_information: List[StateFieldSpec] [Optional]

field private_information: List[StateFieldSpec] [Optional]

field public_information: List[StateFieldSpec] [Optional]

create_state_class() → Type[GameState]

Create a GameState subclass from this configuration using create_model.

pydantic model econagents.adapters.config.StateFieldSpec

Bases: BaseModel

Configuration for a field in the state.

Fields:

• default (Any)

• default_factory (str | None)

• event_key (str | None)

• events (List[str] | None)

• exclude_events (List[str] | None)

• exclude_from_mapping (bool)

• name (str)

• optional (bool)

• type (str)

field name: str [Required]

field type: str [Required]

field default: Any = None

field default_factory: str | None = None

field event_key: str | None = None

field exclude_from_mapping: bool = False

field optional: bool = False

field events: List[str] | None = None

field exclude_events: List[str] | None = None

class econagents.adapters.config.YamlExperimentLoader(config_path: Path)

Bases: object

Load and run an experiment specification from YAML.

load_config() → ExperimentSpec

Load the experiment configuration from the YAML file.

async run_experiment(login_payloads: List[Dict[str, Any]], game_id: int) → None

Run the experiment from this configuration.

Parameters:

login_payloads -- A list of dictionaries containing login information for each agent

async econagents.adapters.config.run_experiment_from_yaml(yaml_path: Path, login_payloads: List[Dict[str, Any]], game_id: int) → None

Run an experiment from a YAML configuration file.

LLM Adapters

class econagents.adapters.llm.BaseLLM

Bases: ABC

Base class for LLM implementations.

observability: ObservabilityProvider = <econagents.adapters.llm.observability.NoOpObservability object>

build_messages(system_prompt: str, user_prompt: str) → list[dict[str, Any]]

Build messages for the LLM.

Parameters:

• system_prompt -- The system prompt for the LLM.

• user_prompt -- The user prompt for the LLM.

Returns:

The messages for the LLM.

abstract async get_response(messages: list[dict[str, Any]], tracing_extra: dict[str, Any], response_schema: Type[BaseModel] | None = None, tools: list[ToolSpec] | None = None, tool_executor: ToolExecutor | None = None, max_tool_iterations: int = 5) → str | BaseModel

Get a response from the LLM.

Parameters:

• messages -- The messages for the LLM.

• tracing_extra -- Extra tracing information passed to observability.

• response_schema -- Optional Pydantic model to use as the structured output schema. Providers that support structured outputs should return a validated instance of this model.

• tools -- Optional provider-agnostic tool specs to advertise. When given together with tool_executor, the adapter runs the native tool-calling loop before returning the final answer.

• tool_executor -- Async callback that executes a single requested tool call and returns its (JSON-serializable) result.

• max_tool_iterations -- Safety cap on tool-call rounds per response.

Returns:

Either a validated response_schema instance or a raw string, depending on provider capabilities and whether a schema was given.

class econagents.adapters.llm.ChatOpenAI(model_name: str = 'gpt-5.4-mini', api_key: str | None = None, reasoning_effort: Literal['minimal', 'low', 'medium', 'high'] | None = None, reasoning_summary: Literal['auto', 'concise', 'detailed'] | None = None, response_kwargs: dict[str, Any] | None = None)

Bases: BaseLLM

OpenAI wrapper built on the Responses API.

Supports structured outputs via a Pydantic response_schema and exposes the reasoning controls available on GPT-5 and other reasoning-capable models. Non-reasoning models should simply leave reasoning_effort and reasoning_summary as None.

async get_response(messages: list[dict[str, Any]], tracing_extra: dict[str, Any], response_schema: Type[BaseModel] | None = None, tools: list[ToolSpec] | None = None, tool_executor: ToolExecutor | None = None, max_tool_iterations: int = 5) → str | BaseModel

Get a response from the OpenAI Responses API.

Parameters:

• messages -- The messages for the LLM.

• tracing_extra -- Extra tracing information passed to observability.

• response_schema -- Optional Pydantic model used as the structured output format. When provided, the method returns a validated instance of the model; otherwise it returns the plain text output from the API.

• tools -- Optional tool specs advertised to the model via the Responses API tools parameter.

• tool_executor -- Async callback used to run each requested tool call.

• max_tool_iterations -- Safety cap on tool-call rounds.

Returns:

A validated response_schema instance, or the raw text output if no schema was provided.

Raises:

ImportError -- If OpenAI is not installed.

class econagents.adapters.llm.LLMProvider(*args, **kwargs)

Bases: Protocol

Interface for model providers used by roles.

async get_response(messages: list[dict[str, Any]], tracing_extra: dict[str, Any], response_schema: Type[BaseModel] | None = None, tools: list[ToolSpec] | None = None, tool_executor: ToolExecutor | None = None, max_tool_iterations: int = 5) → str | BaseModel

Return a raw model response or a validated structured response.

When tools and tool_executor are provided, the adapter runs the provider-native tool-calling loop: it advertises the tools, executes any requested calls via tool_executor, feeds the results back, and repeats up to max_tool_iterations times until the model returns a final answer. The return type is unchanged whether or not tools are used.

build_messages(system_prompt: str, user_prompt: str) → list[dict[str, Any]]

Build provider-specific chat messages.

class econagents.adapters.llm.ObservabilityProvider

Bases: ABC

Base class for observability providers.

abstract track_llm_call(name: str, model: str, messages: List[Dict[str, Any]], response: Any, metadata: Dict[str, Any] | None = None) → None

Track an LLM call.

Parameters:

• name -- Name of the operation.

• model -- Model used for the call.

• messages -- Messages sent to the model.

• response -- Raw response object from the provider SDK.

• metadata -- Additional metadata for the call.

econagents.adapters.llm.get_observability_provider(provider_name: str = 'noop') → ObservabilityProvider

Get an observability provider by name.

Parameters:

provider_name -- The name of the provider to get. Options: "noop", "langsmith", "langfuse"

Returns:

An observability provider.

Raises:

ValueError -- If the provider_name is invalid.

Prompt Adapters

Prompt renderer adapters.

class econagents.adapters.prompts.JinjaPromptRenderer

Bases: object

Render role/phase prompt templates from a directory.

resolve_prompt_file(prompt_type: Literal['system', 'user'], phase: int | str, role: str, prompts_path: Path) → Path | None

Resolve the most specific prompt file for one role.

render(context: dict[str, Any], prompt_type: Literal['system', 'user'], phase: int | str, prompts_path: Path, role_names: list[str], resolver: Callable[[Literal['system', 'user'], int | str, str, Path], Path | None] | None = None, logger: Any | None = None) → str

Render a prompt using role-specific and all-role fallbacks.

Parsing Adapters

Response parser adapters.

class econagents.adapters.parsing.JsonResponseParser

Bases: object

Parse structured-output instances, schema-validated JSON, or raw JSON.

parse(response: str | BaseModel, state: GameStateProtocol, phase: int | str, response_schema: Type[BaseModel] | None = None, logger: Any | None = None) → dict[str, Any]

Parse one LLM response into an action payload.

State Adapters

State projector adapters.

class econagents.adapters.state.EventFieldStateProjector

Bases: object

Apply events through the EventField mapping model.

apply(state: GameState, event: Event) → None

Update the state in place.

Personas

Persona storage and retrieval.

A Persona is a stable identity (demographics, traits, optional bio) stored as a YAML file. Personas are loaded by id from a user-provided directory, falling back to a bundled starter library.

pydantic model econagents.personas.Persona

Bases: BaseModel

Stable, portable identity injected into agent prompts.

Config:

• frozen: bool = True

• extra: str = forbid

Fields:

• bio (str)

• demographics (dict[str, Any])

• id (str)

• traits (dict[str, Any])

field id: str [Required]

field demographics: dict[str, Any] [Optional]

field traits: dict[str, Any] [Optional]

field bio: str = ''

exception econagents.personas.PersonaNotFoundError(persona_id: str, searched: list[Path])

Bases: LookupError

Raised when a persona id cannot be resolved in any configured location.

econagents.personas.load_persona(persona_id: str, user_dir: Path | None = None) → Persona

Resolve persona_id by checking user_dir first, then the bundled library.

If user_dir is not provided, the loader falls back to <cwd>/personas when that directory exists. Pass user_dir explicitly to point at a different location, or just rely on the default when running from a directory that has a personas/ sibling.

Each root is searched recursively for <persona_id>.yaml, so subdirectories (e.g. library/archetypes/) work without the caller knowing about them. Ids must be unique within a tree; if two files share a stem, lookup is deterministic on path-sorted order but a duplicate id is a configuration bug.

Raises PersonaNotFoundError if not found in either location.

econagents.personas.save_persona(persona: Persona, path: Path) → None

Write persona to path as YAML (sorted keys, unicode preserved).