Spaces:
Paused
Paused
File size: 3,356 Bytes
9792ea7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | # -*- coding: utf-8 -*-
"""Base class shared by the team tools."""
from typing import Any, TYPE_CHECKING
from ...permission import (
PermissionBehavior,
PermissionContext,
PermissionDecision,
)
from ...tool import ToolBase
if TYPE_CHECKING:
from ..message_bus import MessageBus
from ..storage import StorageBase
from ..workspace_manager import WorkspaceManagerBase
class _TeamToolBase(ToolBase):
"""Shared base for the team tools.
All team tools are constructed at agent assembly time (in
:func:`get_toolkit`) with the request-scoped ``user_id``,
``session_id``, and ``agent_id`` plus ``storage`` + ``message_bus``
+ ``workspace_manager`` references. Each tool's ``__call__`` does
its work directly via those dependencies — there is no intermediate
service layer.
Permissions: all team tools allow themselves unconditionally — the
agent's authority to call them is already gated by the
role/source-aware logic inside :func:`get_toolkit` that decides
which team tools to attach in the first place.
"""
name: str
description: str
input_schema: dict[str, Any]
is_concurrency_safe: bool = False
is_read_only: bool = True
is_state_injected: bool = False
is_external_tool: bool = False
is_mcp: bool = False
mcp_name: str | None = None
def __init__(
self,
storage: "StorageBase",
message_bus: "MessageBus",
workspace_manager: "WorkspaceManagerBase",
user_id: str,
session_id: str,
agent_id: str,
) -> None:
"""Bind request-scoped identifiers and shared dependencies.
Args:
storage (`StorageBase`):
Application storage.
message_bus (`MessageBus`):
Application message bus for inter-session delivery.
workspace_manager (`WorkspaceManagerBase`):
Workspace manager, consulted by tools that provision
brand-new sessions (e.g. ``AgentInvite``) to honour
the deployment's isolation policy.
user_id (`str`):
The owner user id of the calling agent.
session_id (`str`):
The current session id of the calling agent.
agent_id (`str`):
The id of the agent invoking the tool.
"""
self._storage = storage
self._message_bus = message_bus
self._workspace_manager = workspace_manager
self._user_id = user_id
self._session_id = session_id
self._agent_id = agent_id
async def check_permissions(
self,
tool_input: dict[str, Any],
context: PermissionContext,
) -> PermissionDecision:
"""Always allow — gating is done by tool-attachment logic.
Args:
tool_input (`dict[str, Any]`):
The arguments the agent passed; ignored here.
context (`PermissionContext`):
The active permission context; ignored here.
Returns:
`PermissionDecision`:
An ``ALLOW`` decision with a brief explanation.
"""
return PermissionDecision(
behavior=PermissionBehavior.ALLOW,
message=f"{self.name} is always allowed when attached to the "
f"agent.",
)
|