Proactive agent tools — check/record permissions, queue and execute actions.
These tools are used exclusively by ProactiveAgent to manage the
propose → approve → execute lifecycle for autonomous actions.
Permission key convention: "{action_type}:{context_key}"
Approval response parsing
When the user replies to a pending-actions notification, their message is
expected to contain one or more tokens of the form:
``{action_id} yes`` or ``{action_id} no``
``yes {action_id}`` or ``no {action_id}``
``always yes {action_id}`` → approve + remember
``always no {action_id}`` → deny + remember
``yes all`` / ``no all`` → bulk approve/deny all pending
Call parse_approval_response(text, store) from any channel message handler
to process these replies without running the full agent.
Bases: BaseTool
Look up whether the user has a remembered decision for a permission key.
Source code in src/openjarvis/tools/proactive_tools.py
| def __init__(self, store: Optional[ApprovalStore] = None) -> None:
self._store = store
|
Bases: BaseTool
Queue a proposed action for user approval or immediate execution.
Source code in src/openjarvis/tools/proactive_tools.py
| def __init__(self, store: Optional[ApprovalStore] = None) -> None:
self._store = store
|
Bases: BaseTool
Return all pending (not yet decided) actions as a JSON list.
Source code in src/openjarvis/tools/proactive_tools.py
| def __init__(self, store: Optional[ApprovalStore] = None) -> None:
self._store = store
|
Bases: BaseTool
Record a user approval or denial for a queued action.
Source code in src/openjarvis/tools/proactive_tools.py
| def __init__(self, store: Optional[ApprovalStore] = None) -> None:
self._store = store
|
ExecutePendingActionsTool(store: Optional[ApprovalStore] = None, executor_fn: Optional[Any] = None)
Bases: BaseTool
Execute all approved (or trivial) actions and return a summary.
Source code in src/openjarvis/tools/proactive_tools.py
| def __init__(
self,
store: Optional[ApprovalStore] = None,
executor_fn: Optional[Any] = None,
) -> None:
self._store = store
# executor_fn(action: PendingAction) -> (success: bool, message: str)
self._executor_fn = executor_fn
|
parse_approval_response(text: str, store: Optional[ApprovalStore] = None) -> List[Dict[str, Any]]
Parse a free-text message for approval tokens and update the store.
Returns a list of dicts describing each decision that was processed,
for use in an acknowledgement message back to the user.
Call this from any channel message handler before routing the message
to the main agent, e.g. inside the iMessage daemon or Telegram bot.
Source code in src/openjarvis/tools/proactive_tools.py
| def parse_approval_response(
text: str,
store: Optional[ApprovalStore] = None,
) -> List[Dict[str, Any]]:
"""Parse a free-text message for approval tokens and update the store.
Returns a list of dicts describing each decision that was processed,
for use in an acknowledgement message back to the user.
Call this from any channel message handler before routing the message
to the main agent, e.g. inside the iMessage daemon or Telegram bot.
"""
s = store or get_store()
processed: List[Dict[str, Any]] = []
# Notification template displays ids as `[abc123]`; users naturally reply
# with `{abc123} yes`, `(abc123) yes`, etc. Strip those surrounding
# brackets/braces/parens before regex matching so the word-boundary
# check sees a clean id.
text = re.sub(r"[\[\]\{\}\(\)]", " ", text)
for m in _APPROVAL_RE.finditer(text):
target = (m.group("target") or m.group("target2") or "").lower()
raw_decision = (m.group("decision") or m.group("decision2") or "").lower()
always = bool(m.group("always") or m.group("always2"))
approved = raw_decision in ("yes", "approve")
if target == "all":
pending = s.list_pending()
for action in pending:
new_status = STATUS_APPROVED if approved else STATUS_DENIED
s.update_status(action.id, new_status)
if always and action.tier in (TIER_LOW, TIER_MEDIUM):
decision = (
DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
)
s.set_permission(action.permission_key, decision, approved=approved)
processed.append(
{"id": action.id, "approved": approved, "remembered": always}
)
else:
action = s.get_action(target)
if action is None:
continue
new_status = STATUS_APPROVED if approved else STATUS_DENIED
s.update_status(target, new_status)
remember = always and action.tier in (TIER_LOW, TIER_MEDIUM)
if remember:
decision = DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
s.set_permission(action.permission_key, decision, approved=approved)
processed.append(
{"id": target, "approved": approved, "remembered": remember}
)
return processed
|