trading_agent.gateway.routes.request_routes

Routes for the desks' questions, plans and permission requests to the user.

GET /api/requests the requests, pending ones only unless status=all POST /api/requests/{id}/answer answer a question request POST /api/requests/{id}/plan approve a plan, or ask for changes POST /api/requests/{id}/permission allow a held call once or always, or deny it POST /api/requests/{id}/dismiss close a request without answering GET /api/permissions the standing "Always allow" permissions POST /api/permissions/revoke make a desk ask again before using a tool

Each response is handed to the desk as a new message; the desk never waited for it. A response is recorded as given "via web" unless the client names itself with the header X-Trading-Agent-Client: terminal.

Typical usage example:

application.include_router(RequestRoutes(guard, store, responder).router())

QuestionAnswer

Bases: BaseModel

The answer to one question.

Attributes:

Name	Type	Description
selected	list[str]	The labels of the options chosen.
other	str	The user's own answer typed under "Other", or empty.

AnswerRequest

Bases: BaseModel

The body of an answer to a question request.

Attributes:

Name	Type	Description
answers	list[QuestionAnswer]	One answer per question, in order.

PlanDecisionRequest

Bases: BaseModel

The body of a decision on a plan.

Attributes:

Name	Type	Description
decision	str	"approve" or "revise".
feedback	str	What should change, or a note with an approval.

PermissionDecisionRequest

Bases: BaseModel

The body of a decision on a held call.

Attributes:

Name	Type	Description
decision	str	"allow_once", "always_allow" or "deny".
reason	str	Why it was denied, or a note; may be empty.

RevokeRequest

Bases: BaseModel

The body of a revocation.

Attributes:

Name	Type	Description
desk	str	The desk.
tool_name	str	The tool it must ask about again.

RequestRoutes

RequestRoutes(guard, store, responder, grants=None)

Bases: RouteGroup

The /api/requests routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
guard	AccessGuard	Checks the access token.	required
store	UserRequestStore	Where the requests are kept.	required
responder	UserRequestResponder	Records responses and tells the desks.	required
grants	PermissionGrants | None	The standing permissions, or None when permissions are not used.	None

Source code in src/trading_agent/gateway/routes/request_routes.py

def __init__(
    self,
    guard: access_guard.AccessGuard,
    store: user_request_store_module.UserRequestStore,
    responder: user_request_responder_module.UserRequestResponder,
    grants: permission_grants_module.PermissionGrants | None = None,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        store (user_request_store_module.UserRequestStore): Where the requests are kept.
        responder (user_request_responder_module.UserRequestResponder): Records responses and tells the desks.
        grants (permission_grants_module.PermissionGrants | None): The standing permissions, or None when permissions are not used.
    """
    super().__init__(guard)
    self._store = store
    self._responder = responder
    self._grants = grants

list_requests

list_requests(status='pending')

Lists the requests.

Parameters:

Name	Type	Description	Default
status	str	"pending" for those still waiting, or "all".	'pending'

Returns:

Type	Description
dict[str, Any]	The requests, oldest first.

Source code in src/trading_agent/gateway/routes/request_routes.py

def list_requests(self, status: str = 'pending') -> dict[str, Any]:
    """Lists the requests.

    Args:
        status (str): "pending" for those still waiting, or "all".

    Returns:
        dict[str, Any]: The requests, oldest first.
    """
    if status == 'all':
        found = self._store.all()
    else:
        found = self._store.pending()
    requests = []
    for request in found:
        requests.append(request.to_json_dictionary())
    return {
        'requests': requests,
    }

answer

answer(request_id, body, client=_CLIENT_HEADER)

Answers a question request.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
body	AnswerRequest	The answers.	required
client	str	The client that sent it, from the X-Trading-Agent-Client header.	_CLIENT_HEADER

Returns:

Type	Description
dict[str, Any]	The answered request.

Raises:

Type	Description
HTTPException	No such request (404), it is no longer pending (409), or the answers do not fit it (422).

Source code in src/trading_agent/gateway/routes/request_routes.py

def answer(
    self,
    request_id: str,
    body: AnswerRequest,
    client: str = _CLIENT_HEADER,
) -> dict[str, Any]:
    """Answers a question request.

    Args:
        request_id (str): The request.
        body (AnswerRequest): The answers.
        client (str): The client that sent it, from the X-Trading-Agent-Client header.

    Returns:
        dict[str, Any]: The answered request.

    Raises:
        fastapi.HTTPException: No such request (404), it is no longer pending (409), or the answers do not fit it (422).
    """
    answers = []
    for answer in body.answers:
        answers.append(answer.model_dump())
    return self._respond(
        request_id,
        lambda: self._responder.answer_question(
            request_id, answers, self._via(client)
        ),
    )

decide_plan

decide_plan(request_id, body, client=_CLIENT_HEADER)

Approves a plan or asks for changes.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
body	PlanDecisionRequest	The decision and feedback.	required
client	str	The client that sent it, from the X-Trading-Agent-Client header.	_CLIENT_HEADER

Returns:

Type	Description
dict[str, Any]	The decided request.

Raises:

Type	Description
HTTPException	No such request (404), it is no longer pending (409), or the decision is not valid (422).

Source code in src/trading_agent/gateway/routes/request_routes.py

def decide_plan(
    self,
    request_id: str,
    body: PlanDecisionRequest,
    client: str = _CLIENT_HEADER,
) -> dict[str, Any]:
    """Approves a plan or asks for changes.

    Args:
        request_id (str): The request.
        body (PlanDecisionRequest): The decision and feedback.
        client (str): The client that sent it, from the X-Trading-Agent-Client header.

    Returns:
        dict[str, Any]: The decided request.

    Raises:
        fastapi.HTTPException: No such request (404), it is no longer pending (409), or the decision is not valid (422).
    """
    return self._respond(
        request_id,
        lambda: self._responder.decide_plan(
            request_id,
            body.decision,
            body.feedback,
            self._via(client),
        ),
    )

decide_permission async

decide_permission(request_id, body, client=_CLIENT_HEADER)

Allows a held call once or always, running it, or denies it.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
body	PermissionDecisionRequest	The decision and reason.	required
client	str	The client that sent it, from the X-Trading-Agent-Client header.	_CLIENT_HEADER

Returns:

Type	Description
dict[str, Any]	The decided request, with the call's result when it ran.

Raises:

Type	Description
HTTPException	No such request (404), it is no longer pending or has lapsed (409), or the decision is not valid (422).

Source code in src/trading_agent/gateway/routes/request_routes.py

async def decide_permission(
    self,
    request_id: str,
    body: PermissionDecisionRequest,
    client: str = _CLIENT_HEADER,
) -> dict[str, Any]:
    """Allows a held call once or always, running it, or denies it.

    Args:
        request_id (str): The request.
        body (PermissionDecisionRequest): The decision and reason.
        client (str): The client that sent it, from the X-Trading-Agent-Client header.

    Returns:
        dict[str, Any]: The decided request, with the call's result when it ran.

    Raises:
        fastapi.HTTPException: No such request (404), it is no longer pending or has lapsed (409), or the decision is not valid (422).
    """
    try:
        self._store.get(request_id)
    except user_request_store_module.UserRequestError as error:
        raise fastapi.HTTPException(status_code=404, detail=str(error)) from error
    try:
        request = await self._responder.decide_permission(
            request_id,
            body.decision,
            body.reason,
            self._via(client),
        )
    except user_request_store_module.UserRequestError as error:
        raise fastapi.HTTPException(status_code=409, detail=str(error)) from error
    except ValueError as error:
        raise fastapi.HTTPException(status_code=422, detail=str(error)) from error
    return request.to_json_dictionary()

list_permissions

list_permissions()

Lists the standing permissions.

Returns:

Type	Description
dict[str, Any]	The grants, oldest first.

Source code in src/trading_agent/gateway/routes/request_routes.py

def list_permissions(self) -> dict[str, Any]:
    """Lists the standing permissions.

    Returns:
        dict[str, Any]: The grants, oldest first.
    """
    grants = []
    if self._grants is not None:
        for grant in self._grants.all():
            grants.append(
                {
                    'desk': grant.desk,
                    'tool_name': grant.tool_name,
                    'granted_at': grant.granted_at,
                }
            )
    return {
        'grants': grants,
    }

revoke_permission

revoke_permission(body)

Makes a desk ask again before using a tool.

Parameters:

Name	Type	Description	Default
body	RevokeRequest	The desk and the tool.	required

Returns:

Type	Description
dict[str, Any]	Whether a grant was removed.

Raises:

Type	Description
HTTPException	There is no such grant (404).

Source code in src/trading_agent/gateway/routes/request_routes.py

def revoke_permission(self, body: RevokeRequest) -> dict[str, Any]:
    """Makes a desk ask again before using a tool.

    Args:
        body (RevokeRequest): The desk and the tool.

    Returns:
        dict[str, Any]: Whether a grant was removed.

    Raises:
        fastapi.HTTPException: There is no such grant (404).
    """
    if self._grants is None or not self._grants.revoke(body.desk, body.tool_name):
        raise fastapi.HTTPException(
            status_code=404,
            detail=f'No standing permission for {body.tool_name!r} on desk {body.desk!r}',
        )
    return {
        'revoked': True,
    }

dismiss

dismiss(request_id, client=_CLIENT_HEADER)

Closes a request without answering it.

Parameters:

Name	Type	Description	Default
request_id	str	The request.	required
client	str	The client that sent it, from the X-Trading-Agent-Client header.	_CLIENT_HEADER

Returns:

Type	Description
dict[str, Any]	The dismissed request.

Raises:

Type	Description
HTTPException	No such request (404), or it is no longer pending (409).

Source code in src/trading_agent/gateway/routes/request_routes.py

def dismiss(
    self,
    request_id: str,
    client: str = _CLIENT_HEADER,
) -> dict[str, Any]:
    """Closes a request without answering it.

    Args:
        request_id (str): The request.
        client (str): The client that sent it, from the X-Trading-Agent-Client header.

    Returns:
        dict[str, Any]: The dismissed request.

    Raises:
        fastapi.HTTPException: No such request (404), or it is no longer pending (409).
    """
    return self._respond(
        request_id,
        lambda: self._responder.dismiss(request_id, self._via(client)),
    )