Context

The Context object is a type-safe shared state container that persists conversation state across all Rounds within a Session, providing centralized access to logs, costs, application state, and execution metadata.

Quick Reference:

Overview

The Context object serves as the central state store for sessions:

  1. Type Safety: Enum-based attribute names with type definitions
  2. Default Values: Automatic initialization with sensible defaults
  3. Auto-Syncing: Current round values sync automatically
  4. Serialization: Convert to/from dict for persistence
  5. Dispatcher Attachment: Command execution integration

Architecture

graph TB subgraph "Context Container" CTX[Context Dataclass] VALUES[Attribute Values Dict] NAMES[ContextNames Enum] end subgraph "Access Patterns" GET[get method] SET[set method] UPDATE[update_dict method] TO_DICT[to_dict method] FROM_DICT[from_dict method] end subgraph "Auto-Sync Properties" PROP_STEP[current_round_step] PROP_COST[current_round_cost] PROP_SUBTASK[current_round_subtask_amount] end subgraph "Shared Across" SESS[Session] R1[Round 1] R2[Round 2] R3[Round 3] AGENTHost[HostAgent] AGENTApp[AppAgent] end CTX --> VALUES NAMES --> VALUES GET --> VALUES SET --> VALUES UPDATE --> VALUES PROP_STEP -.auto-updates.-> VALUES PROP_COST -.auto-updates.-> VALUES PROP_SUBTASK -.auto-updates.-> VALUES SESS -.shares.-> CTX R1 -.shares.-> CTX R2 -.shares.-> CTX R3 -.shares.-> CTX AGENTHost -.reads/writes.-> CTX AGENTApp -.reads/writes.-> CTX style CTX fill:#e1f5ff style VALUES fill:#fff4e1 style PROP_STEP fill:#f0ffe1 style SESS fill:#ffe1f5

ContextNames Enum

All context attributes are defined in the ContextNames enum for type safety:

from ufo.module.context import ContextNames

# Type-safe attribute names
request = context.get(ContextNames.REQUEST)
context.set(ContextNames.SESSION_COST, 0.42)

Attribute Categories

30+ Context Attributes

Context attributes are organized into 7 logical categories.

1. Identifiers & Mode

Context attributes for session and mode identification.

Attribute Type Default Description
ID int 0 Session ID
MODE str "" Execution mode (normal, service, etc.)
CURRENT_ROUND_ID int 0 Current round number

2. Execution State

Attribute Type Default Description
REQUEST str "" Current user request
SUBTASK str "" Current subtask for AppAgent
PREVIOUS_SUBTASKS List [] Previous subtasks history
HOST_MESSAGE List [] HostAgent → AppAgent messages
ROUND_RESULT str "" Current round result

3. Cost Tracking

Attribute Type Default Description
SESSION_COST float 0.0 Total session cost ($)
ROUND_COST Dict[int, float] {} Cost per round
CURRENT_ROUND_COST float 0.0 Current round cost (auto-sync)

4. Step Counting

Attribute Type Default Description
SESSION_STEP int 0 Total steps in session
ROUND_STEP Dict[int, int] {} Steps per round
CURRENT_ROUND_STEP int 0 Current round steps (auto-sync)
ROUND_SUBTASK_AMOUNT Dict[int, int] {} Subtasks per round
CURRENT_ROUND_SUBTASK_AMOUNT int 0 Current subtasks (auto-sync)

5. Application Context

Attribute Type Default Description
APPLICATION_WINDOW UIAWrapper None Current application window
APPLICATION_WINDOW_INFO Any - Window metadata
APPLICATION_PROCESS_NAME str "" Process name (e.g., "WINWORD.EXE")
APPLICATION_ROOT_NAME str "" Root UI element name
CONTROL_REANNOTATION List [] Control re-annotations

6. Logging

Attribute Type Default Description
LOG_PATH str "" Log directory path
LOGGER Logger None Session logger
REQUEST_LOGGER Logger None LLM request logger
EVALUATION_LOGGER Logger None Evaluation logger
STRUCTURAL_LOGS defaultdict defaultdict(...) Structured logs

7. Tools & Communication

Attribute Type Default Description
TOOL_INFO Dict {} Available tools metadata
DEVICE_INFO List [] Connected device information (Galaxy)
CONSTELLATION TaskConstellation None Task constellation (Galaxy)
WEAVING_MODE WeavingMode CREATION Weaving mode (Galaxy)

Complete Attribute Reference

All 30+ attributes with types and defaults.

class ContextNames(Enum):
    # Identifiers
    ID = "ID"                                        # int, default: 0
    MODE = "MODE"                                    # str, default: ""
    CURRENT_ROUND_ID = "CURRENT_ROUND_ID"            # int, default: 0

    # Requests & Tasks
    REQUEST = "REQUEST"                              # str, default: ""
    SUBTASK = "SUBTASK"                              # str, default: ""
    PREVIOUS_SUBTASKS = "PREVIOUS_SUBTASKS"          # List, default: []
    HOST_MESSAGE = "HOST_MESSAGE"                    # List, default: []
    ROUND_RESULT = "ROUND_RESULT"                    # str, default: ""

    # Costs
    SESSION_COST = "SESSION_COST"                    # float, default: 0.0
    ROUND_COST = "ROUND_COST"                        # Dict, default: {}
    CURRENT_ROUND_COST = "CURRENT_ROUND_COST"        # float, default: 0.0

    # Steps
    SESSION_STEP = "SESSION_STEP"                    # int, default: 0
    ROUND_STEP = "ROUND_STEP"                        # Dict, default: {}
    CURRENT_ROUND_STEP = "CURRENT_ROUND_STEP"        # int, default: 0
    ROUND_SUBTASK_AMOUNT = "ROUND_SUBTASK_AMOUNT"    # Dict, default: {}
    CURRENT_ROUND_SUBTASK_AMOUNT = "CURRENT_ROUND_SUBTASK_AMOUNT"  # int, default: 0

    # Application
    APPLICATION_WINDOW = "APPLICATION_WINDOW"        # UIAWrapper, default: None
    APPLICATION_WINDOW_INFO = "APPLICATION_WINDOW_INFO"  # Any
    APPLICATION_PROCESS_NAME = "APPLICATION_PROCESS_NAME"  # str, default: ""
    APPLICATION_ROOT_NAME = "APPLICATION_ROOT_NAME"  # str, default: ""
    CONTROL_REANNOTATION = "CONTROL_REANNOTATION"    # List, default: []

    # Logging
    LOG_PATH = "LOG_PATH"                            # str, default: ""
    LOGGER = "LOGGER"                                # Logger, default: None
    REQUEST_LOGGER = "REQUEST_LOGGER"                # Logger, default: None
    EVALUATION_LOGGER = "EVALUATION_LOGGER"          # Logger, default: None
    STRUCTURAL_LOGS = "STRUCTURAL_LOGS"              # defaultdict

    # Tools & Devices
    TOOL_INFO = "TOOL_INFO"                          # Dict, default: {}
    DEVICE_INFO = "DEVICE_INFO"                      # List, default: []
    CONSTELLATION = "CONSTELLATION"                  # TaskConstellation, default: None
    WEAVING_MODE = "WEAVING_MODE"                    # WeavingMode, default: CREATION

Context Methods

get()

Retrieve a value from context:

def get(self, name: ContextNames, default: Any = None) -> Any

Example:

request = context.get(ContextNames.REQUEST)
# Returns "" if not set

cost = context.get(ContextNames.SESSION_COST, 0.0)
# Returns 0.0 if not set or uses provided default

set()

Set a context value:

def set(self, name: ContextNames, value: Any) -> None

Example:

context.set(ContextNames.REQUEST, "Send an email to John")
context.set(ContextNames.SESSION_COST, 0.42)
context.set(ContextNames.APPLICATION_PROCESS_NAME, "WINWORD.EXE")

update_dict()

Batch update multiple values:

def update_dict(self, updates: Dict[ContextNames, Any]) -> None

Example:

context.update_dict({
    ContextNames.REQUEST: "New task",
    ContextNames.MODE: "normal",
    ContextNames.SESSION_STEP: 10
})

to_dict()

Serialize context to dictionary:

def to_dict(self) -> Dict[str, Any]

Returns: Dictionary with only JSON-serializable values

Example:

context_dict = context.to_dict()
# Save to file
json.dump(context_dict, open("context.json", "w"))

Excluded from serialization: - Loggers (LOGGER, REQUEST_LOGGER, EVALUATION_LOGGER) - Window objects (APPLICATION_WINDOW) - Non-serializable objects

from_dict()

Restore context from dictionary:

@staticmethod
def from_dict(data: Dict[str, Any]) -> "Context"

Example:

# Load from file
data = json.load(open("context.json"))
context = Context.from_dict(data)

attach_command_dispatcher()

Attach dispatcher for command execution:

def attach_command_dispatcher(self, dispatcher: BasicCommandDispatcher) -> None

Example:

from ufo.module.dispatcher import LocalCommandDispatcher

dispatcher = LocalCommandDispatcher(session, mcp_manager)
context.attach_command_dispatcher(dispatcher)

# Now rounds can execute commands via context

Auto-Syncing Properties

These properties automatically sync with current round values in dictionaries.

current_round_step

@property
def current_round_step(self) -> int:
    """Get current round step."""
    return self.attributes.get(ContextNames.ROUND_STEP, {}).get(
        self.attributes.get(ContextNames.CURRENT_ROUND_ID, 0), 0
    )

@current_round_step.setter
def current_round_step(self, value: int) -> None:
    """Set current round step and update dict."""
    round_id = self.attributes.get(ContextNames.CURRENT_ROUND_ID, 0)
    self.attributes[ContextNames.ROUND_STEP][round_id] = value
    self.attributes[ContextNames.CURRENT_ROUND_STEP] = value

Usage:

# Reading
steps = context.current_round_step

# Writing (updates both ROUND_STEP dict and CURRENT_ROUND_STEP)
context.current_round_step = 5

current_round_cost

Auto-syncs cost tracking:

# Reading
cost = context.current_round_cost

# Writing (updates both ROUND_COST dict and CURRENT_ROUND_COST)
context.current_round_cost += 0.01

current_round_subtask_amount

Auto-syncs subtask counting:

# Reading
subtasks = context.current_round_subtask_amount

# Writing
context.current_round_subtask_amount += 1

Usage Patterns

Pattern 1: Session Initialization

from ufo.module.context import Context, ContextNames

# Create context
context = Context()

# Initialize session metadata
context.set(ContextNames.ID, 0)
context.set(ContextNames.MODE, "normal")
context.set(ContextNames.LOG_PATH, "./logs/task_001/")
context.set(ContextNames.REQUEST, "Send an email")

Pattern 2: Round Execution

# At round start
context.set(ContextNames.CURRENT_ROUND_ID, round_id)

# During round
context.current_round_step += 1
context.current_round_cost += agent_cost

# Agent reads state
request = context.get(ContextNames.REQUEST)
process_name = context.get(ContextNames.APPLICATION_PROCESS_NAME)

Pattern 3: Cost Tracking

# Agent incurs cost
agent_cost = llm_call_cost()
context.current_round_cost += agent_cost

# Session total auto-updates
context.set(
    ContextNames.SESSION_COST,
    context.get(ContextNames.SESSION_COST, 0.0) + agent_cost
)

# Print summary
print(f"Round cost: ${context.current_round_cost:.4f}")
print(f"Session total: ${context.get(ContextNames.SESSION_COST):.4f}")

Pattern 4: Application Tracking

# Agent selects application
context.set(ContextNames.APPLICATION_PROCESS_NAME, "WINWORD.EXE")
context.set(ContextNames.APPLICATION_ROOT_NAME, "Document1 - Word")
context.set(ContextNames.APPLICATION_WINDOW, word_window)

# Later rounds access same app
app_window = context.get(ContextNames.APPLICATION_WINDOW)
if app_window:
    app_window.set_focus()

Pattern 5: Logging

# Setup loggers
context.set(ContextNames.LOGGER, session_logger)
context.set(ContextNames.REQUEST_LOGGER, request_logger)

# Use throughout session
logger = context.get(ContextNames.LOGGER)
logger.info("Round started")

request_logger = context.get(ContextNames.REQUEST_LOGGER)
request_logger.log_request(prompt, response)

Pattern 6: Persistence

# Save context state
context_dict = context.to_dict()
with open("checkpoint.json", "w") as f:
    json.dump(context_dict, f, indent=2)

# Resume from checkpoint
with open("checkpoint.json") as f:
    data = json.load(f)
restored_context = Context.from_dict(data)

Best Practices

Type Safety

Use Enum Names

Always use ContextNames enum instead of strings:

# ✅ Good
context.get(ContextNames.REQUEST)

# ❌ Bad
context.attributes["REQUEST"]

Default Values

Leverage Defaults

ContextNames provides sensible defaults:

# No need to check for None
cost = context.get(ContextNames.SESSION_COST)  # Returns 0.0 if unset

# Explicit default
steps = context.get(ContextNames.SESSION_STEP, 0)

Auto-Sync

Use Auto-Sync Properties

For current round values, use auto-sync properties:

# ✅ Good - auto-syncs both dicts
context.current_round_cost += 0.01

# ❌ Manual - must update both
round_id = context.get(ContextNames.CURRENT_ROUND_ID)
context.attributes[ContextNames.ROUND_COST][round_id] += 0.01
context.attributes[ContextNames.CURRENT_ROUND_COST] += 0.01

Reference

Context Dataclass

The context class that maintains the context for the session and agent.

current_round_cost property writable

Get the current round cost.

current_round_step property writable

Get the current round step.

current_round_subtask_amount property writable

Get the current round subtask index.

add_to_structural_logs(data)

Add data to the structural logs.

Parameters:
  • data (Dict[str, Any]) –

    The data to add to the structural logs.
Source code in module/context.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def add_to_structural_logs(self, data: Dict[str, Any]) -> None:
    """
    Add data to the structural logs.
    :param data: The data to add to the structural logs.
    """

    round_key = data.get("Round", None)
    subtask_key = data.get("SubtaskIndex", None)

    if round_key is None or subtask_key is None:
        return

    remaining_items = {key: data[key] for key in data}
    self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][subtask_key].append(
        remaining_items
    )

attach_command_dispatcher(command_dispatcher)

Attach a command dispatcher to the context.

Parameters:
  • command_dispatcher (BasicCommandDispatcher) –

    The command dispatcher to attach.
Source code in module/context.py 
391
392
393
394
395
396
397
398
def attach_command_dispatcher(
    self, command_dispatcher: BasicCommandDispatcher
) -> None:
    """
    Attach a command dispatcher to the context.
    :param command_dispatcher: The command dispatcher to attach.
    """
    self.command_dispatcher = command_dispatcher

filter_structural_logs(round_key, subtask_key, keys)

Filter the structural logs.

Parameters:
  • round_key (int) –

    The round key.

  • subtask_key (int) –

    The subtask key.

  • keys (Union[str, List[str]]) –

    The keys to filter.
Returns:
  • Union[List[Any], List[Dict[str, Any]]] –

    The filtered structural logs.
Source code in module/context.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def filter_structural_logs(
    self, round_key: int, subtask_key: int, keys: Union[str, List[str]]
) -> Union[List[Any], List[Dict[str, Any]]]:
    """
    Filter the structural logs.
    :param round_key: The round key.
    :param subtask_key: The subtask key.
    :param keys: The keys to filter.
    :return: The filtered structural logs.
    """

    structural_logs = self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][
        subtask_key
    ]

    if isinstance(keys, str):
        return [log[keys] for log in structural_logs]
    elif isinstance(keys, list):
        return [{key: log[key] for key in keys} for log in structural_logs]
    else:
        raise TypeError(f"Keys should be a string or a list of strings.")

from_dict(context_dict)

Load the context from a dictionary.

Parameters:
  • context_dict (Dict[str, Any]) –

    The dictionary of the context.
Source code in module/context.py 
379
380
381
382
383
384
385
386
387
388
389
def from_dict(self, context_dict: Dict[str, Any]) -> None:
    """
    Load the context from a dictionary.
    :param context_dict: The dictionary of the context.
    """
    for key in ContextNames:
        if key.name in context_dict:
            self._context[key.name] = context_dict.get(key.name)

    # Sync the current round step and cost
    self._sync_round_values()

get(key)

Get the value from the context.

Parameters:
Returns:
  • Any –

    The value from the context.
Source code in module/context.py 
207
208
209
210
211
212
213
214
215
def get(self, key: ContextNames) -> Any:
    """
    Get the value from the context.
    :param key: The context name.
    :return: The value from the context.
    """
    # Sync the current round step and cost
    self._sync_round_values()
    return self._context.get(key.name)

set(key, value)

Set the value in the context.

Parameters:
  • key (ContextNames) –

    The context name.

  • value (Any) –

    The value to set in the context.
Source code in module/context.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def set(self, key: ContextNames, value: Any) -> None:
    """
    Set the value in the context.
    :param key: The context name.
    :param value: The value to set in the context.
    """
    if key.name in self._context:
        self._context[key.name] = value
        # Sync the current round step and cost
        if key == ContextNames.CURRENT_ROUND_STEP:
            self.current_round_step = value
        if key == ContextNames.CURRENT_ROUND_COST:
            self.current_round_cost = value
        if key == ContextNames.CURRENT_ROUND_SUBTASK_AMOUNT:
            self.current_round_subtask_amount = value
    else:
        raise KeyError(f"Key '{key}' is not a valid context name.")

to_dict(ensure_serializable=False)

Convert the context to a dictionary.

Parameters:
  • ensure_serializable (bool, default: False ) –

    Ensure the context is serializable.
Returns:
  • Dict[str, Any] –

    The dictionary of the context.
Source code in module/context.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def to_dict(self, ensure_serializable: bool = False) -> Dict[str, Any]:
    """
    Convert the context to a dictionary.
    :param ensure_serializable: Ensure the context is serializable.
    :return: The dictionary of the context.
    """

    import copy

    context_dict = copy.deepcopy(self._context)

    if ensure_serializable:

        for key in ContextNames:
            if key.name in context_dict:
                logger.warning(
                    f"The value of Context.{key.name} is not serializable."
                )
                if not is_json_serializable(context_dict[key.name]):

                    context_dict[key.name] = None

    return context_dict

update_dict(key, value)

Add a dictionary to a context key. The value and the context key should be dictionaries.

Parameters:
  • key (ContextNames) –

    The context key to update.

  • value (Dict[str, Any]) –

    The dictionary to add to the context key.
Source code in module/context.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def update_dict(self, key: ContextNames, value: Dict[str, Any]) -> None:
    """
    Add a dictionary to a context key. The value and the context key should be dictionaries.
    :param key: The context key to update.
    :param value: The dictionary to add to the context key.
    """
    if key.name in self._context:
        context_value = self._context[key.name]
        if isinstance(value, dict) and isinstance(context_value, dict):
            self._context[key.name].update(value)
        else:
            raise TypeError(
                f"Value for key '{key.name}' is {key.value}, requires a dictionary."
            )
    else:
        raise KeyError(f"Key '{key.name}' is not a valid context name.")

ContextNames Enum

Bases: Enum

The context names.

default_value property

Get the default value for the context name based on its type.

Returns:
  • Any –

    The default value for the context name.

type property

Get the type of the context name.

Returns:
  • Type –

    The type of the context name.

See Also

  • Session - Session lifecycle and context usage
  • Round - Round execution with context
  • Overview - Module system architecture