angee.agents_runtime_pydantic.toolsets

Compose REBAC-gated native and remote toolsets for one agent session.

Built-in Angee tools execute directly through their registered FastMCP Tool.run implementation. FastMCP remains the registration/compiler owner; the in-process adapter only translates its definitions and results into pydantic-ai's toolset contract. Remote servers keep pydantic-ai's MCP transport.

MAX_TOOL_RESULT_CHARS

Maximum inline characters returned from one in-process tool call.

TOOL_RESULT_TRUNCATED

Marker appended when an in-process tool result exceeds the inline ceiling.

ALWAYS_LOADED_TOOLS

Small native core kept visible beside pydantic-ai's own search_tools.

ToolGrantAccess

@dataclass
class ToolGrantAccess()

Share one advertisement lookup and fresh per-call checks across toolsets.

granted_ids

async def granted_ids() -> frozenset[str] | None

Return accessible qualified ids, or None for a universal admin grant.

check

async def check(server_sqid: str, tool_name: str) -> None

Re-gate one invocation so revocation beats stale advertisement.

AngeeToolset

@dataclass
class AngeeToolset(AbstractToolset[Any])

Execute granted built-in FastMCP tools natively for one persisted session.

id

@property
def id() -> str

Return the stable id of the process-native toolset.

get_tools

async def get_tools(ctx: Any) -> dict[str, ToolsetTool[Any]]

Advertise the one accessible grant set intersected with the live registry.

call_tool

async def call_tool(name: str, tool_args: dict[str, Any], ctx: Any,
                    tool: ToolsetTool[Any]) -> Any

Re-gate, select the execution actor, run directly, and bound the result.

ToolGrantToolset

@dataclass
class ToolGrantToolset(WrapperToolset[Any])

Apply the same REBAC grant gate to an external MCP toolset.

get_tools

async def get_tools(ctx: Any) -> dict[str, ToolsetTool[Any]]

Filter remote advertisement through the shared use grant set.

call_tool

async def call_tool(name: str, tool_args: dict[str, Any], ctx: Any,
                    tool: ToolsetTool[Any]) -> Any

Re-gate immediately before delegating to the remote MCP transport.

toolsets_for_session

def toolsets_for_session(session: Any) -> list[Any]

Return native plus remote toolsets authorized for session.