API
econagents: A Python library that lets you use LLM agents in economic experiments.
- class econagents.AgentManager(url: str | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, logger: Logger | None = None)[source]
Bases:
LoggerMixinAgent Manager for handling connections, message routing, and event handling.
The AgentManager provides a high-level interface for connecting to a server, sending messages, and routing received messages to appropriate handlers. It also supports pre- and post-event hooks for intercepting and processing messages.
Connection parameters (URL and authentication mechanism) can be:
Provided at initialization time
Injected later using property setters:
manager.url = "wss://example.com/ws"
manager.auth_mechanism = SimpleLoginPayloadAuth()
manager.auth_mechanism_kwargs = {"username": "user", "password": "pass"}
This delayed injection pattern allows for more flexible configuration and testing.
- Parameters:
url (Optional[str]) -- WebSocket URL to connect to
auth_mechanism (Optional[AuthenticationMechanism]) -- Authentication mechanism
auth_mechanism_kwargs (Optional[dict[str, Any]]) -- Keyword arguments to pass to auth_mechanism
logger (Optional[logging.Logger]) -- Logger instance
- property url: str | None
Get the WebSocket URL.
- property auth_mechanism: AuthenticationMechanism | None
Get the authentication mechanism.
- property auth_mechanism_kwargs: dict[str, Any] | None
Get the authentication mechanism keyword arguments.
- property end_game_event_type: str
Get the event type that triggers the agent to stop.
- async on_message(message: Message)[source]
Default implementation to handle incoming messages from the server.
For event-type messages, routes them to on_event. Subclasses can override this method for custom handling.
- Parameters:
message (Message) -- Incoming message from the server
- async send_message(message: str)[source]
Send a message through the transport layer.
- Parameters:
message (str) -- Message to send
- async on_event(message: Message)[source]
Handle event messages by routing to specific handlers.
The execution flow is:
Global pre-event hooks
Event-specific pre-event hooks
Global event handlers
Event-specific handlers
Event-specific post-event hooks
Global post-event hooks
Subclasses can override this method for custom event handling.
- Parameters:
message (Message) -- Incoming event message from the server
- register_event_handler(event_type: str, handler: Callable[[Message], Any])[source]
Register a handler function for a specific event type.
- Parameters:
event_type (str) -- The type of event to handle
handler (Callable[[Message], Any]) -- Function that takes a Message object and handles the event
- register_global_event_handler(handler: Callable[[Message], Any])[source]
Register a handler function for all events.
- Parameters:
handler (Callable[[Message], Any]) -- Function that takes a Message object and handles any event
- register_pre_event_hook(event_type: str, hook: Callable[[Message], Any])[source]
Register a hook to execute before handlers for a specific event type.
- Parameters:
event_type (str) -- The type of event to hook
hook (Callable[[Message], Any]) -- Function that takes a Message object and runs before handlers
- register_global_pre_event_hook(hook: Callable[[Message], Any])[source]
Register a hook to execute before handlers for all events.
- Parameters:
hook (Callable[[Message], Any]) -- Function that takes a Message object and runs before any handlers
- register_post_event_hook(event_type: str, hook: Callable[[Message], Any])[source]
Register a hook to execute after handlers for a specific event type.
- Parameters:
event_type (str) -- The type of event to hook
hook (Callable[[Message], Any]) -- Function that takes a Message object and runs after handlers
- register_global_post_event_hook(hook: Callable[[Message], Any])[source]
Register a hook to execute after handlers for all events.
- Parameters:
hook (Callable[[Message], Any]) -- Function that takes a Message object and runs after all handlers
- unregister_event_handler(event_type: str, handler: Callable | None = None)[source]
Unregister handler(s) for a specific event type.
- Parameters:
event_type (str) -- The type of event
handler (Optional[Callable]) -- Optional handler to remove. If None, removes all handlers for this event type.
- unregister_global_event_handler(handler: Callable | None = None)[source]
Unregister global event handler(s).
- Parameters:
handler (Optional[Callable]) -- Optional handler to remove. If None, removes all global handlers.
- unregister_pre_event_hook(event_type: str, hook: Callable | None = None)[source]
Unregister pre-event hook(s) for a specific event type.
- Parameters:
event_type (str) -- The type of event
hook (Optional[Callable]) -- Optional hook to remove. If None, removes all pre-event hooks for this event type.
- unregister_global_pre_event_hook(hook: Callable | None = None)[source]
Unregister global pre-event hook(s).
- Parameters:
hook (Optional[Callable]) -- Optional hook to remove. If None, removes all global pre-event hooks.
- unregister_post_event_hook(event_type: str, hook: Callable | None = None)[source]
Unregister post-event hook(s) for a specific event type.
- Parameters:
event_type (str) -- The type of event
hook (Optional[Callable]) -- Optional hook to remove. If None, removes all post-event hooks for this event type.
- class econagents.AgentRole(logger: Logger | None = None, persona: Persona | None = None)[source]
Bases:
ABC,Generic[StateT_contra],LoggerMixinBase agent 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 template-based prompts and allows customization of behavior 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: BaseLLM
Language model instance for generating responses
- task_phases: ClassVar[list[int]] = []
List of phases this agent should participate in (empty means all phases)
- task_phases_excluded: ClassVar[list[int]] = []
Alternative way to specify phases this agent should participate in, listed phases are excluded (empty means nothing excluded)
- response_schemas: ClassVar[Dict[int, 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.
- auto_render_persona: ClassVar[bool] = True
When
Trueand a persona is attached, append a standard markdown block describing the persona to the end of the system prompt. Set toFalseto take full control via{{ persona }}in your own template.
- render_prompt(context: dict, prompt_type: Literal['system', 'user'], phase: int, prompts_path: Path) str[source]
Render a prompt template with the given context.
Template resolution order:
Role-specific phase prompt (e.g., "role_name_system_phase_1.jinja2")
Role-specific general prompt (e.g., "role_name_system.jinja2")
All-role phase prompt (e.g., "all_system_phase_1.jinja2")
All-role general prompt (e.g., "all_system.jinja2")
- Parameters:
context (dict) -- Template context variables
prompt_type (Literal["system", "user"]) -- 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, handler: Callable[[StateT_contra], str]) None[source]
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, handler: Callable[[StateT_contra], str]) None[source]
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, parser: Callable[[str | BaseModel, StateT_contra], dict]) None[source]
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, schema: Type[BaseModel]) None[source]
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) Type[BaseModel] | None[source]
Return the schema to use for a given phase, if any.
- register_phase_handler(phase: int, handler: Callable[[int, StateT_contra], Any]) None[source]
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[source]
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[source]
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[source]
Parse the LLM response for the current phase.
Resolution order:
A phase-specific parser registered via
register_response_parser.If the provider returned a validated Pydantic instance, its
model_dump().If a response schema is registered for this phase (or a default schema is set), validate the raw string against it.
Fall back to
json.loadson 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, state: StateT_contra, prompts_path: Path) dict | None[source]
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, state: StateT_contra, prompts_path: Path) dict | None[source]
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]
- class econagents.BaseConfigParser(config_path: Path)[source]
Bases:
objectBase configuration parser with no custom event handlers.
- create_manager(game_id: int, state: GameState, agent_role: AgentRole | None, auth_kwargs: Dict[str, Any]) PhaseManager[source]
Create a manager instance based on the configuration. This base implementation has no custom event handlers.
- Parameters:
game_id -- The game ID
state -- The game state instance
agent_role -- The agent role instance
auth_kwargs -- Authentication mechanism keyword arguments
- Returns:
A PhaseManager instance
- class econagents.BasicConfigParser(config_path: Path)[source]
Bases:
BaseConfigParserBasic configuration parser that adds a custom event handler for sending a player-is-ready message when it receives a certain message from the server.
- create_manager(game_id: int, state: GameState, agent_role: AgentRole | None, auth_kwargs: Dict[str, Any]) PhaseManager[source]
Create a manager instance with a custom event handler for the assign-name event.
- Parameters:
game_id -- The game ID
state -- The game state instance
agent_role -- The agent role instance
auth_kwargs -- Authentication mechanism keyword arguments
- Returns:
A PhaseManager instance with custom event handlers
- 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[source]
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[PhaseManager])[source]
Bases:
object- get_agent_logger(agent_id: int, game_id: int) Logger[source]
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[source]
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[source]
Clean up logging resources, stopping all queue listeners. Should be called when shutting down the game runner.
- async spawn_agent(agent_manager: PhaseManager, agent_id: int) None[source]
Spawn an agent and connect it to the game.
- Parameters:
agent_manager (PhaseManager) -- Agent manager to spawn
agent_id (int) -- Agent identifier
- pydantic model econagents.GameState[source]
Bases:
BaseModelGame state for a given game
- Fields:
meta (econagents.core.state.game.MetaInformation)private_information (econagents.core.state.game.PrivateInformation)public_information (econagents.core.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[source]
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
- pydantic model econagents.HybridGameRunnerConfig[source]
Bases:
GameRunnerConfigConfiguration class for TurnBasedGameRunner.
- Fields:
continuous_phases (list[int])max_action_delay (int)min_action_delay (int)
- field continuous_phases: list[int] [Optional]
- field min_action_delay: int = 5
- field max_action_delay: int = 10
- class econagents.HybridPhaseManager(continuous_phases: set[int] | None = None, url: str | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, phase_transition_event: str | None = None, phase_identifier_key: str | None = None, min_action_delay: int | None = None, max_action_delay: int | None = None, state: GameState | None = None, agent_role: AgentRole | None = None, logger: Logger | None = None, prompts_dir: Path | None = None)[source]
Bases:
PhaseManagerA manager for games that combine turn-based and continuous action phases.
This manager extends PhaseManager and configures it with specific phases that should be treated as continuous. By default, all phases are treated as turn-based unless explicitly included in the continuous_phases parameter.
For continuous-time phases, the manager will automatically execute actions periodically with random delays between min_action_delay and max_action_delay seconds.
- Parameters:
continuous_phases (Optional[set[int]]) -- Set of phase numbers that should be treated as continuous
url (Optional[str]) -- WebSocket server URL
auth_mechanism (Optional[AuthenticationMechanism]) -- Authentication mechanism to use
auth_mechanism_kwargs (Optional[dict[str, Any]]) -- Keyword arguments for the authentication mechanism
phase_transition_event (Optional[str]) -- Event name for phase transitions
phase_identifier_key (Optional[str]) -- Key in the event data that identifies the phase
min_action_delay (Optional[int]) -- Minimum delay in seconds between actions in continuous-time phases
max_action_delay (Optional[int]) -- Maximum delay in seconds between actions in continuous-time phases
state (Optional[GameState]) -- Game state object to track game state
agent_role (Optional[AgentRole]) -- Agent role instance to handle game phases
logger (Optional[logging.Logger]) -- Logger instance for tracking events
prompts_dir (Optional[Path]) -- Directory containing the prompt templates
- pydantic model econagents.MetaInformation[source]
Bases:
BaseModelMeta information for the game
- Config:
extra: str = allow
arbitrary_types_allowed: bool = False
- Fields:
game_id (int)phase (int)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 = 0
Current phase of the game
- class econagents.PhaseManager(url: str | None = None, phase_transition_event: str | None = None, phase_identifier_key: str | None = None, continuous_phases: set[int] | None = None, min_action_delay: int | None = None, max_action_delay: int | None = None, state: GameState | None = None, agent_role: AgentRole | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, logger: Logger | None = None, prompts_dir: Path | None = None)[source]
Bases:
AgentManager,ABCAbstract manager that handles the concept of 'phases' in a game.
This manager standardizes the interface for phase-based games with optional continuous-time phase handling.
Features: 1. Standardized interface for starting a phase
Optional continuous "tick loop" for phases
All configuration parameters can be:
Provided at initialization time
Injected later using property setters
- Parameters:
url (Optional[str]) -- WebSocket server URL
phase_transition_event (Optional[str]) -- Event name for phase transitions
phase_identifier_key (Optional[str]) -- Key in the event data that identifies the phase
continuous_phases (Optional[set[int]]) -- set of phase numbers that should be treated as continuous
min_action_delay (Optional[int]) -- Minimum delay in seconds between actions in continuous-time phases
max_action_delay (Optional[int]) -- Maximum delay in seconds between actions in continuous-time phases
state (Optional[GameState]) -- Game state object to track game state
agent_role (Optional[AgentRole]) -- Agent role instance to handle game phases
auth_mechanism (Optional[AuthenticationMechanism]) -- Authentication mechanism to use
auth_mechanism_kwargs (Optional[dict[str, Any]]) -- Keyword arguments for the authentication mechanism
logger (Optional[logging.Logger]) -- Logger instance for tracking events
prompts_dir (Optional[Path]) -- Directory containing the prompt templates
- property phase_transition_event: str
Get the phase transition event name.
- property phase_identifier_key: str
Get the phase identifier key.
- property continuous_phases: set[int]
Get the set of continuous-time phases.
- property min_action_delay: int
Get the minimum action delay.
- property max_action_delay: int
Get the maximum action delay.
- property prompts_dir: Path
Get the prompts directory.
- property llm_provider
Get the LLM provider from the agent role.
- async handle_phase_transition(new_phase: int | None)[source]
Handle a phase transition.
This method is the main orchestrator for phase transitions: 1. If leaving a continuous-time phase, stops the continuous task 2. Updates the current phase 3. Starts a continuous task if entering a continuous-time phase 4. Executes a single action if entering a non-continuous-time phase
- Parameters:
new_phase (Optional[int]) -- The new phase number
- pydantic model econagents.PrivateInformation[source]
Bases:
BaseModelPrivate information for each agent in the game
- Config:
extra: str = allow
arbitrary_types_allowed: bool = False
- pydantic model econagents.PublicInformation[source]
Bases:
BaseModelPublic information for the game
- Config:
extra: str = allow
arbitrary_types_allowed: bool = False
- pydantic model econagents.TurnBasedGameRunnerConfig[source]
Bases:
GameRunnerConfigConfiguration class for TurnBasedGameRunner.
- Fields:
- class econagents.TurnBasedPhaseManager(url: str | None = None, phase_transition_event: str | None = None, phase_identifier_key: str | None = None, auth_mechanism: AuthenticationMechanism | None = None, auth_mechanism_kwargs: dict[str, Any] | None = None, state: GameState | None = None, agent_role: AgentRole | None = None, logger: Logger | None = None, prompts_dir: Path | None = None)[source]
Bases:
PhaseManagerA manager for turn-based games that handles phase transitions.
This manager inherits from PhaseManager and provides a concrete implementation for executing actions in each phase. All phases are treated as turn-based, meaning actions are only taken when explicitly triggered (no continuous actions).
- Parameters:
url (Optional[str]) -- WebSocket server URL
phase_transition_event (Optional[str]) -- Event name for phase transitions
phase_identifier_key (Optional[str]) -- Key in the event data that identifies the phase
auth_mechanism (Optional[AuthenticationMechanism]) -- Authentication mechanism to use
auth_mechanism_kwargs (Optional[dict[str, Any]]) -- Keyword arguments for the authentication mechanism
state (Optional[GameState]) -- Game state object to track game state
agent_role (Optional[AgentRole]) -- Agent role instance to handle game phases
logger (Optional[logging.Logger]) -- Logger instance for tracking events
prompts_dir (Optional[Path]) -- Directory containing the prompt templates
- 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)[source]
Bases:
LoggerMixinResponsible for connecting to a WebSocket, sending/receiving messages, and reporting received messages to a callback function.
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[source]
Bases:
BaseModelStable, portable identity injected into agent prompts.
- Config:
frozen: bool = True
extra: str = forbid
- Fields:
- 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])[source]
Bases:
LookupErrorRaised when a persona id cannot be resolved in any configured location.
- econagents.personas.load_persona(persona_id: str, user_dir: Path | None = None) Persona[source]
Resolve
persona_idby checkinguser_dirfirst, then the bundled library.If
user_diris not provided, the loader falls back to<cwd>/personaswhen that directory exists. Passuser_direxplicitly to point at a different location, or just rely on the default when running from a directory that has apersonas/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
PersonaNotFoundErrorif not found in either location.