trading_agent.requests.user_request_responder

Records the user's response to a desk's request and hands it to the desk as a new message.

The web page and Telegram both answer through this class, so a request answered in one place shows as answered in the other.

Typical usage example:

responder = UserRequestResponder(store, event_bus, desks) responder.answer_question(request_id, [{'selected': ['Wait'], 'other': ''}], 'web')

UserRequestResponder

UserRequestResponder(store, event_bus, desks, tool_registries=None, grants=None)

Answers questions, decides plans and permissions, and dismisses requests on the user's behalf.

Creates the responder.

Parameters:

Name	Type	Description	Default
store	UserRequestStore	Where requests are kept.	required
event_bus	EventBus	Where each response is published.	required
desks	Mapping[str, DeskRuntime]	The running desks, keyed by name, which receive the responses.	required
tool_registries	Mapping[str, ToolRegistry] | None	Each desk's tools, which run a call the user allows; None when permissions are not used.	None
grants	PermissionGrants | None	Where "Always allow" is saved; None when permissions are not used.	None

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

def __init__(
    self,
    store: user_request_store_module.UserRequestStore,
    event_bus: event_bus_module.EventBus,
    desks: Mapping[str, desk_runtime.DeskRuntime],
    tool_registries: Mapping[str, tool_registry_module.ToolRegistry] | None = None,
    grants: permission_grants_module.PermissionGrants | None = None,
):
    """Creates the responder.

    Args:
        store (user_request_store_module.UserRequestStore): Where requests are kept.
        event_bus (event_bus_module.EventBus): Where each response is published.
        desks (Mapping[str, desk_runtime.DeskRuntime]): The running desks, keyed by name, which receive the responses.
        tool_registries (Mapping[str, tool_registry_module.ToolRegistry] | None): Each desk's tools, which run a call the user allows; None when permissions are not used.
        grants (permission_grants_module.PermissionGrants | None): Where "Always allow" is saved; None when permissions are not used.
    """
    self._store = store
    self._event_bus = event_bus
    self._desks = desks
    self._tool_registries = tool_registries or {}
    self._grants = grants

answer_question

answer_question(request_id, answers, via)

Records the answers to a question request and tells the desk.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
answers	Sequence[dict[str, Any]]	One {"selected": [labels], "other": text} per question.	required
via	str	"web", "terminal" or "telegram".	required

Returns:

Type	Description
UserRequest	The answered request.

Raises:

Type	Description
UserRequestError	The request does not exist or is no longer pending.
ValueError	The request is not a question, or the answers do not fit it.

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

def answer_question(
    self,
    request_id: str,
    answers: Sequence[dict[str, Any]],
    via: str,
) -> user_request_module.UserRequest:
    """Records the answers to a question request and tells the desk.

    Args:
        request_id (str): The request.
        answers (Sequence[dict[str, Any]]): One {"selected": [labels], "other": text} per question.
        via (str): "web", "terminal" or "telegram".

    Returns:
        user_request_module.UserRequest: The answered request.

    Raises:
        user_request_store_module.UserRequestError: The request does not exist or is no longer pending.
        ValueError: The request is not a question, or the answers do not fit it.
    """
    request = self._store.pending_request(request_id)
    if request.kind != user_request_module.QUESTION:
        raise ValueError(f'Request {request_id} is a {request.kind}, not a question')
    response = question_response_module.QuestionResponse(request, answers)
    response.check()
    return self._finish(
        request,
        user_request_module.ANSWERED,
        response.stored(),
        response.message(),
        via,
    )

decide_plan

decide_plan(request_id, decision, feedback, via)

Records the decision on a plan and tells the desk.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
decision	str	"approve" or "revise".	required
feedback	str	What should change, or a note with an approval.	required
via	str	"web", "terminal" or "telegram".	required

Returns:

Type	Description
UserRequest	The decided request.

Raises:

Type	Description
UserRequestError	The request does not exist or is no longer pending.
ValueError	The request is not a plan, or the decision is not valid.

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

def decide_plan(
    self,
    request_id: str,
    decision: str,
    feedback: str,
    via: str,
) -> user_request_module.UserRequest:
    """Records the decision on a plan and tells the desk.

    Args:
        request_id (str): The request.
        decision (str): "approve" or "revise".
        feedback (str): What should change, or a note with an approval.
        via (str): "web", "terminal" or "telegram".

    Returns:
        user_request_module.UserRequest: The decided request.

    Raises:
        user_request_store_module.UserRequestError: The request does not exist or is no longer pending.
        ValueError: The request is not a plan, or the decision is not valid.
    """
    request = self._store.pending_request(request_id)
    if request.kind != user_request_module.PLAN:
        raise ValueError(f'Request {request_id} is a {request.kind}, not a plan')
    response = plan_response_module.PlanResponse(request, decision, feedback)
    response.check()
    return self._finish(
        request,
        user_request_module.ANSWERED,
        response.stored(),
        response.message(),
        via,
    )

decide_permission async

decide_permission(request_id, decision, reason, via)

Records the decision on a held call, runs the call when it is allowed, and tells the desk.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
decision	str	"allow_once", "always_allow" or "deny".	required
reason	str	Why it was denied, or a note; may be empty.	required
via	str	"web", "terminal" or "telegram".	required

Returns:

Type	Description
UserRequest	The decided request.

Raises:

Type	Description
UserRequestError	The request does not exist, is no longer pending, or has expired.
ValueError	The request is not a permission request, the decision is not valid, or the desk's tools are not known.

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

async def decide_permission(
    self,
    request_id: str,
    decision: str,
    reason: str,
    via: str,
) -> user_request_module.UserRequest:
    """Records the decision on a held call, runs the call when it is allowed, and tells the desk.

    Args:
        request_id (str): The request.
        decision (str): "allow_once", "always_allow" or "deny".
        reason (str): Why it was denied, or a note; may be empty.
        via (str): "web", "terminal" or "telegram".

    Returns:
        user_request_module.UserRequest: The decided request.

    Raises:
        user_request_store_module.UserRequestError: The request does not exist, is no longer pending, or has expired.
        ValueError: The request is not a permission request, the decision is not valid, or the desk's tools are not known.
    """
    request = self._store.pending_request(request_id)
    if request.kind != user_request_module.PERMISSION:
        raise ValueError(f'Request {request_id} is a {request.kind}, not a permission request')
    response = permission_response_module.PermissionResponse(request, decision, reason)
    response.check()
    result = None
    if response.allows:
        registry = self._tool_registries.get(request.desk)
        if registry is None:
            raise ValueError(f'The tools of desk {request.desk!r} are not known here')
        if decision == permission_response_module.ALWAYS_ALLOW and self._grants is not None:
            self._grants.grant(request.desk, request.content['tool_name'])
        call = messages.ToolCall(
            call_id=str(request.content.get('call_id', request.request_id)),
            name=request.content['tool_name'],
            arguments=dict(request.content.get('arguments') or {}),
        )
        result = await registry.run(call)
    return self._finish(
        request,
        user_request_module.ANSWERED,
        response.stored(result),
        response.message(result),
        via,
    )

report_expired

report_expired(request)

Tells the desk that a request lapsed without an answer.

Parameters:

Name	Type	Description	Default
request	UserRequest	The request, already saved as expired.	required

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

def report_expired(self, request: user_request_module.UserRequest) -> None:
    """Tells the desk that a request lapsed without an answer.

    Args:
        request (user_request_module.UserRequest): The request, already saved as expired.
    """
    if request.kind == user_request_module.PERMISSION:
        message = f'Your request to run {request.title()} (request {request.request_id}) lapsed without an answer, so it was not run. If it still matters, decide again with fresh prices.'
    else:
        message = f'Your {request.kind} "{request.title()}" (request {request.request_id}) lapsed without an answer. Use your own judgment within the rules.'
    self._publish_response(request, user_request_module.EXPIRED, {}, '')
    self._send(request, message)

dismiss

dismiss(request_id, via)

Closes a request without answering it, and tells the desk to use its own judgment.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
via	str	"web", "terminal" or "telegram".	required

Returns:

Type	Description
UserRequest	The dismissed request.

Raises:

Type	Description
UserRequestError	The request does not exist or is no longer pending.

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

def dismiss(self, request_id: str, via: str) -> user_request_module.UserRequest:
    """Closes a request without answering it, and tells the desk to use its own judgment.

    Args:
        request_id (str): The request.
        via (str): "web", "terminal" or "telegram".

    Returns:
        user_request_module.UserRequest: The dismissed request.

    Raises:
        user_request_store_module.UserRequestError: The request does not exist or is no longer pending.
    """
    request = self._store.pending_request(request_id)
    message = f'The user closed your {request.kind} "{request.title()}" (request {request.request_id}) without answering. Use your own judgment within the rules, or ask again later if it still matters.'
    return self._finish(
        request,
        user_request_module.DISMISSED,
        {},
        message,
        via,
    )