trading_agent.requests.user_request

One thing a desk asked the user: a question, a plan to approve, or permission for an action.

A desk never waits for the answer. The request is saved and shown to the user, the desk's turn ends, and the user's answer reaches the desk later as a new message.

Typical usage example:

request = UserRequest.new('main', QUESTION, {'questions': [...]}) request.to_json_dictionary()

QUESTION module-attribute

QUESTION = 'question'

PLAN module-attribute

PLAN = 'plan'

PERMISSION module-attribute

PERMISSION = 'permission'

KINDS module-attribute

KINDS = (QUESTION, PLAN, PERMISSION)

PENDING module-attribute

PENDING = 'pending'

ANSWERED module-attribute

ANSWERED = 'answered'

DISMISSED module-attribute

DISMISSED = 'dismissed'

EXPIRED module-attribute

EXPIRED = 'expired'

UserRequest dataclass

UserRequest(request_id, desk, kind, content, created_at, status, response=None, responded_at=None, responded_via=None, expires_at=None)

A question, plan or permission request, and the user's response once given.

Attributes:

Name	Type	Description
request_id	str	Its identifier.
desk	str	The desk that asked.
kind	str	"question", "plan" or "permission".
content	dict[str, Any]	What was asked: the questions, the plan, or the action awaiting permission.
created_at	str	When it was asked, as ISO 8601 text in UTC.
status	str	"pending", "answered", "dismissed" or "expired".
response	dict[str, Any] | None	The user's response, or None while pending.
responded_at	str | None	When the user responded, as ISO 8601 text, or None.
responded_via	str | None	Where the user responded, "web", "terminal" or "telegram", or None.
expires_at	str | None	When the request lapses, as ISO 8601 text, or None when it never does.

new classmethod

new(desk, kind, content, expires_at=None)

Creates a pending request with a new identifier.

Parameters:

Name	Type	Description	Default
desk	str	The desk asking.	required
kind	str	"question", "plan" or "permission".	required
content	dict[str, Any]	What is asked.	required
expires_at	datetime | None	When it lapses, or None for never.	None

Returns:

Type	Description
UserRequest	The request.

Raises:

Type	Description
ValueError	The kind is not one of KINDS.

Source code in src/trading_agent/requests/user_request.py

@classmethod
def new(
    cls,
    desk: str,
    kind: str,
    content: dict[str, Any],
    expires_at: datetime.datetime | None = None,
) -> Self:
    """Creates a pending request with a new identifier.

    Args:
        desk (str): The desk asking.
        kind (str): "question", "plan" or "permission".
        content (dict[str, Any]): What is asked.
        expires_at (datetime.datetime | None): When it lapses, or None for never.

    Returns:
        UserRequest: The request.

    Raises:
        ValueError: The kind is not one of KINDS.
    """
    if kind not in KINDS:
        raise ValueError(f'Unknown request kind: {kind!r}')
    now = datetime.datetime.now(datetime.UTC)
    return cls(
        request_id=f'{now:%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:6]}',
        desk=desk,
        kind=kind,
        content=content,
        created_at=now.isoformat(),
        status=PENDING,
        expires_at=None if expires_at is None else expires_at.isoformat(),
    )

from_json_dictionary classmethod

from_json_dictionary(dictionary)

Reads a request saved by to_json_dictionary.

Parameters:

Name	Type	Description	Default
dictionary	dict[str, Any]	The saved fields.	required

Returns:

Type	Description
UserRequest	The request.

Source code in src/trading_agent/requests/user_request.py

@classmethod
def from_json_dictionary(cls, dictionary: dict[str, Any]) -> Self:
    """Reads a request saved by to_json_dictionary.

    Args:
        dictionary (dict[str, Any]): The saved fields.

    Returns:
        UserRequest: The request.
    """
    return cls(**dictionary)

to_json_dictionary

to_json_dictionary()

Converts the request to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	Every field.

Source code in src/trading_agent/requests/user_request.py

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the request to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: Every field.
    """
    return dataclasses.asdict(self)

title

title()

A short name for the request, for messages and notifications.

Returns:

Type	Description
str	The plan's title, the action's name, or the first question.

Source code in src/trading_agent/requests/user_request.py

def title(self) -> str:
    """A short name for the request, for messages and notifications.

    Returns:
        str: The plan's title, the action's name, or the first question.
    """
    if self.kind == PLAN:
        return str(self.content.get('title', 'plan'))
    if self.kind == PERMISSION:
        return str(self.content.get('tool_name', 'an action'))
    questions = self.content.get('questions') or [
        {},
    ]
    return str(questions[0].get('question', 'question'))

is_expired

is_expired(now)

Says whether a pending request has passed its expiry time.

Parameters:

Name	Type	Description	Default
now	datetime	The current time, with a time zone.	required

Returns:

Type	Description
bool	True when it has an expiry time and that time has passed.

Source code in src/trading_agent/requests/user_request.py

def is_expired(self, now: datetime.datetime) -> bool:
    """Says whether a pending request has passed its expiry time.

    Args:
        now (datetime.datetime): The current time, with a time zone.

    Returns:
        bool: True when it has an expiry time and that time has passed.
    """
    if self.expires_at is None:
        return False
    return now >= datetime.datetime.fromisoformat(self.expires_at)