Tools

Tools infrastructure for building Agents.

Resolution chain for tool imports: 1. Core tools (always available — defined directly in this module) 2. parrot_tools (ai-parrot-tools installed package) 3. plugins.tools (user/deploy-time plugin directory) 4. TOOL_REGISTRY (declarative registry from ai-parrot-tools) 5. Legacy dynamic_import_helper (backward-compat submodule resolution)

Submodule redirector

from parrot.tools.prophetforecast import X is transparently redirected to from parrot_tools.prophetforecast import X when no local submodule exists. This is done via a sys.meta_path finder installed at import time.

AbstractTool

AbstractTool(name: Optional[str] = None, description: Optional[str] = None, output_dir: Optional[Union[str, Path]] = None, base_url: Optional[str] = None, static_dir: Optional[Union[str, Path]] = None, routing_meta: Optional[Dict] = None, executor: Optional[AbstractToolExecutor] = None, webhook_callback_url: Optional[str] = None, remote_timeout_seconds: int = 300, **kwargs)

Bases: EventEmitterMixin, ABC

Abstract base class for all tools in the ai-parrot framework.

This class provides a unified interface for tools that can be used by both conversational bots and agents. It includes common functionality like: - Name and description management - JSON schema generation - File path management - Logging and error handling - Async/sync execution support - Lifecycle event emission (FEAT-176)

Initialize the tool.

PARAMETER	DESCRIPTION
name	Tool name (defaults to class name) TYPE: Optional[str] DEFAULT: None
description	Tool description TYPE: Optional[str] DEFAULT: None
output_dir	Directory for output files (if tool generates files) TYPE: Optional[Union[str, Path]] DEFAULT: None
base_url	Base URL for serving static files TYPE: Optional[str] DEFAULT: None
static_dir	Static directory path TYPE: Optional[Union[str, Path]] DEFAULT: None
routing_meta	Optional routing hints dict for CapabilityRegistry. Supported keys: "description", "not_for". TYPE: Optional[Dict] DEFAULT: None
executor	Optional :class:AbstractToolExecutor that runs the tool off-process (Kubernetes pod, Qworker, etc.). When None (the default), the tool executes in-process — identical to the legacy behaviour. TYPE: Optional[AbstractToolExecutor] DEFAULT: None
webhook_callback_url	When set, the executor returns a ToolResult(status="pending") immediately and the final result arrives at this URL out-of-band. Only honoured by executors that support async delivery. TYPE: Optional[str] DEFAULT: None
remote_timeout_seconds	Max wall-clock seconds the executor waits for the remote runtime to return. Ignored when executor is None. TYPE: int DEFAULT: 300
**kwargs	Additional configuration DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def __init__(
    self,
    name: Optional[str] = None,
    description: Optional[str] = None,
    output_dir: Optional[Union[str, Path]] = None,
    base_url: Optional[str] = None,
    static_dir: Optional[Union[str, Path]] = None,
    routing_meta: Optional[Dict] = None,
    executor: Optional["AbstractToolExecutor"] = None,
    webhook_callback_url: Optional[str] = None,
    remote_timeout_seconds: int = 300,
    **kwargs
):
    """
    Initialize the tool.

    Args:
        name: Tool name (defaults to class name)
        description: Tool description
        output_dir: Directory for output files (if tool generates files)
        base_url: Base URL for serving static files
        static_dir: Static directory path
        routing_meta: Optional routing hints dict for CapabilityRegistry.
            Supported keys: ``"description"``, ``"not_for"``.
        executor: Optional :class:`AbstractToolExecutor` that runs the
            tool off-process (Kubernetes pod, Qworker, etc.). When
            ``None`` (the default), the tool executes in-process —
            identical to the legacy behaviour.
        webhook_callback_url: When set, the executor returns a
            ``ToolResult(status="pending")`` immediately and the
            final result arrives at this URL out-of-band. Only
            honoured by executors that support async delivery.
        remote_timeout_seconds: Max wall-clock seconds the executor
            waits for the remote runtime to return. Ignored when
            ``executor`` is ``None``.
        **kwargs: Additional configuration
    """
    # routing_meta — per-instance to avoid shared mutable default
    self.routing_meta: Dict = routing_meta if routing_meta is not None else {}

    # Remote execution wiring (None = legacy in-process behaviour)
    self.executor: Optional["AbstractToolExecutor"] = executor
    self.webhook_callback_url: Optional[str] = webhook_callback_url
    self.remote_timeout_seconds: int = int(remote_timeout_seconds)

    # Store initialization parameters for cloning. The live executor
    # instance is captured so clone() reuses the same transport;
    # ToolExecutionEnvelope serialization strips it before sending.
    self._init_kwargs = {
        'name': name,
        'description': description,
        'output_dir': output_dir,
        'base_url': base_url,
        'static_dir': static_dir,
        'executor': executor,
        'webhook_callback_url': webhook_callback_url,
        'remote_timeout_seconds': remote_timeout_seconds,
        **kwargs
    }

    # Set name and description
    self.name = name or self.name or self.__class__.__name__
    self.description = description or self.__class__.__doc__ or f"Tool: {self.name}"

    # Initialize permission context (per-call, set in execute())
    self._current_pctx: Optional[Any] = None

    # Set up logging
    self.logger = logging.getLogger(
        f'{self.name}.Tool'
    )

    # JSON encoders/decoders
    self._json_encoder = json_encoder
    self._json_decoder = json_decoder
    self._json = JSONContent()

    # File and URL configuration
    self.base_url = base_url or BASE_STATIC_URL
    self.static_url = base_url or BASE_STATIC_URL
    parsed = urlparse(self.static_url)
    self._base_scheme_netloc = (parsed.scheme, parsed.netloc)

    # Set up directories
    self.static_dir = Path(static_dir or STATIC_DIR).resolve()

    self.output_dir = Path(output_dir).resolve() if output_dir else self._default_output_dir()

    # Ensure output directory exists if specified
    if self.output_dir and not self.output_dir.exists():
        self.output_dir.mkdir(parents=True, exist_ok=True)

    # FEAT-176: initialise per-instance event registry
    self._init_events()

clone

clone()

Create a new instance of this tool with the same configuration.

This method creates a new instance of the tool class with all the initialization parameters that were passed to the current instance. Subclasses can override _get_clone_kwargs() to customize which parameters are cloned and which are not.

RETURNS	DESCRIPTION
	New instance of the same tool class with cloned configuration

Example

dbtool = DatabaseTool(connection_string="postgresql://...") new_tool = dbtool.clone()

new_tool is a fresh instance with the same configuration

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def clone(self):
    """
    Create a new instance of this tool with the same configuration.

    This method creates a new instance of the tool class with all the
    initialization parameters that were passed to the current instance.
    Subclasses can override _get_clone_kwargs() to customize which
    parameters are cloned and which are not.

    Returns:
        New instance of the same tool class with cloned configuration

    Example:
        >>> dbtool = DatabaseTool(connection_string="postgresql://...")
        >>> new_tool = dbtool.clone()
        >>> # new_tool is a fresh instance with the same configuration
    """
    clone_kwargs = self._get_clone_kwargs()
    return self.__class__(**clone_kwargs)

get_schema

get_schema() -> Dict[str, Any]

Get the JSON schema for this tool.

RETURNS	DESCRIPTION
Dict[str, Any]	JSON schema dictionary compatible with LLM tool registration

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def get_schema(self) -> Dict[str, Any]:
    """
    Get the JSON schema for this tool.

    Returns:
        JSON schema dictionary compatible with LLM tool registration
    """

    def _enforce_no_extra_fields(definition: Any) -> None:
        """Recursively set ``additionalProperties`` to ``False`` for objects.

        OpenAI tools require every object schema (including nested ones) to
        explicitly disallow extra properties. Pydantic's generated schema only
        sets this flag at the top level, so we walk the entire schema tree and
        ensure every object definition is strict.
        """

        if not isinstance(definition, dict):
            return

        if definition.get("type") == "object":
            definition.setdefault("properties", {})
            definition.setdefault("additionalProperties", False)

        # Recurse into common schema containers
        for key in ("properties", "patternProperties"):
            if isinstance(definition.get(key), dict):
                for sub in definition[key].values():
                    _enforce_no_extra_fields(sub)

        for key in ("items", "additionalItems"):
            _enforce_no_extra_fields(definition.get(key))

        for key in ("anyOf", "oneOf", "allOf"):
            if isinstance(definition.get(key), list):
                for sub in definition[key]:
                    _enforce_no_extra_fields(sub)

        # Handle $defs/definitions used by Pydantic
        for key in ("$defs", "definitions"):
            if isinstance(definition.get(key), dict):
                for sub in definition[key].values():
                    _enforce_no_extra_fields(sub)

    schema = {
        "name": self.name,
        "description": self.description,
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
            "additionalProperties": False
        }
    }

    # If args_schema is defined, use it to build the parameters
    if self.args_schema and self.args_schema != AbstractToolArgsSchema:
        pydantic_schema = self.args_schema.model_json_schema()
        properties = pydantic_schema.get("properties", {})
        required = pydantic_schema.get("required", [])

        # Strip context fields (injected at runtime, not by the LLM)
        ctx = getattr(self.args_schema, '_context_fields', frozenset())
        if ctx:
            properties = {
                k: v for k, v in properties.items() if k not in ctx
            }
            required = [r for r in required if r not in ctx]

        schema["parameters"] = {
            "type": "object",
            "properties": properties,
            "required": required,
            "additionalProperties": False,
            "$defs": pydantic_schema.get("$defs", {}),
        }

    _enforce_no_extra_fields(schema["parameters"])
    return schema

get_tool_schema

get_tool_schema() -> Dict[str, Any]

Get the JSON schema for the tool's arguments. Alias for get_schema() for backward compatibility.

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary containing the JSON schema

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def get_tool_schema(self) -> Dict[str, Any]:
    """
    Get the JSON schema for the tool's arguments.
    Alias for get_schema() for backward compatibility.

    Returns:
        Dictionary containing the JSON schema
    """
    return self.get_schema()

validate_args

validate_args(**kwargs) -> BaseModel

Validate arguments using the tool's schema.

PARAMETER	DESCRIPTION
**kwargs	Arguments to validate DEFAULT: {}

RETURNS	DESCRIPTION
BaseModel	Validated arguments as Pydantic model instance

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def validate_args(self, **kwargs) -> BaseModel:
    """
    Validate arguments using the tool's schema.

    Args:
        **kwargs: Arguments to validate

    Returns:
        Validated arguments as Pydantic model instance
    """
    if not self.args_schema or self.args_schema == AbstractToolArgsSchema:
        # If no schema is defined, return a basic model with the kwargs
        return AbstractToolArgsSchema()
    try:
        kwargs = self._coerce_json_strings(kwargs)
        result = self.args_schema(**kwargs)
        if not result:
            self.logger.warning(
                "Validation failed for %s with args: %s", self.name, kwargs
            )
        return result
    except Exception as e:
        self.logger.error("Validation error in %s: %s", self.name, e)
        raise ValueError(
            f"Invalid arguments for {self.name}: {e}"
        ) from e

execute async

execute(*args, **kwargs) -> ToolResult

Execute the tool with error handling and result standardization.

PARAMETER	DESCRIPTION
*args	Positional arguments for the tool. DEFAULT: ()
**kwargs	Tool arguments. Special kwargs are: - _permission_context: PermissionContext for Layer 2 enforcement - _resolver: AbstractPermissionResolver for permission checks DEFAULT: {}

RETURNS	DESCRIPTION
ToolResult	Standardized ToolResult. Returns status='forbidden' if permission denied.

TODO: Use the Global Registry to share data between tools.

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

async def execute(self, *args, **kwargs) -> ToolResult:
    """
    Execute the tool with error handling and result standardization.

    Args:
        *args: Positional arguments for the tool.
        **kwargs: Tool arguments. Special kwargs are:
            - _permission_context: PermissionContext for Layer 2 enforcement
            - _resolver: AbstractPermissionResolver for permission checks

    Returns:
        Standardized ToolResult. Returns status='forbidden' if permission denied.

    TODO: Use the Global Registry to share data between tools.
    """
    # ── Permission check (Layer 2 safety net) ────────────────────────────
    pctx = kwargs.pop('_permission_context', None)
    resolver = kwargs.pop('_resolver', None)

    # FEAT-264: credential broker kwargs (never enter LLM-visible args_schema)
    _broker: Optional["CredentialBroker"] = kwargs.pop('_broker', None)
    _cred_channel: str = kwargs.pop('_cred_channel', 'unknown')
    _cred_user_id: Optional[str] = kwargs.pop('_cred_user_id', None)

    if pctx is not None and resolver is not None:
        required = getattr(self, '_required_permissions', set())
        allowed = await resolver.can_execute(pctx, self.name, required)
        if not allowed:
            self.logger.warning(
                "Permission denied: user=%s tool=%s required=%s",
                pctx.user_id, self.name, required
            )
            return ToolResult(
                success=False,
                status='forbidden',
                result=None,
                error=f"Permission denied: '{self.name}' requires {required}",
                metadata={
                    "tool_name": self.name,
                    "user_id": pctx.user_id,
                    "required_permissions": list(required),
                }
            )

    # Store for lifecycle hooks.  ToolkitTool._execute reads ``_current_pctx``
    # and injects it back into the ``_pre_execute`` / ``_post_execute`` calls
    # so toolkits can access the request context (FEAT-107 oauth2_3lo mode).
    # NOTE: _current_pctx is intentionally NOT guarded by a lock.
    # This design assumes single-agent sessions where a given ToolkitTool
    # instance is never awaited concurrently from multiple coroutines.
    # If that assumption changes, replace this with a contextvars.ContextVar.
    self._current_pctx = pctx

    # ── FEAT-176: lifecycle — derive / mint trace context ─────────────────
    _tool_name = self.name or type(self).__name__
    parent_tc = pctx.trace_context if pctx is not None else None
    tool_tc: TraceContext = parent_tc.child() if parent_tc is not None else TraceContext.new_root()
    # Propagate updated trace to sub-agents: write back BEFORE _execute so
    # any AgentB invoked inside the tool sees tool_tc as its parent span.
    if pctx is not None:
        pctx.trace_context = tool_tc

    # Emit BeforeToolCallEvent (sync — low-overhead hot path)
    self.events.emit_nowait(BeforeToolCallEvent(
        trace_context=tool_tc,
        tool_name=_tool_name,
        tool_class=type(self).__name__,
        args_summary=self._args_summary(kwargs),
        source_type="tool",
        source_name=_tool_name,
    ))
    _lc_t0 = time.perf_counter()

    # ── Normal execution ─────────────────────────────────────────────────
    try:
        self.logger.info("Executing tool: %s", self.name)

        # Validate arguments
        validated_args = self.validate_args(**kwargs)

        # Resolve the kwargs dict that the tool actually receives.
        if hasattr(validated_args, 'model_dump'):
            resolved_kwargs = validated_args.model_dump()
        else:
            resolved_kwargs = dict(kwargs)

        # ── FEAT-264: credential seam ─────────────────────────────────────
        # Gate is active only when the tool declares credential_provider
        # AND a broker is available.  Tools without credential_provider
        # (the vast majority) are unaffected — this branch is never entered.
        _cred_token = _CREDENTIAL_VAR.set(None)  # establish reset point
        try:
            if self.credential_provider and _broker is not None:
                from ..auth.credentials import NeedsAuth, CredentialRequired
                from ..auth.broker import CredentialBroker  # noqa: F401

                if not _cred_user_id:
                    raise ValueError(
                        f"Tool {self.name!r} requires credential_provider="
                        f"{self.credential_provider!r} but no user identity was "
                        "provided by the caller (fail closed — no service identity)."
                    )

                resolution = await _broker.resolve(
                    self.credential_provider,
                    _cred_channel,
                    _cred_user_id,
                    tool_name=self.name,
                )

                if isinstance(resolution, NeedsAuth):
                    raise CredentialRequired(
                        resolution.provider,
                        resolution.auth_url,
                        resolution.auth_kind,
                    )

                # Hit — inject credential onto per-call ContextVar
                _CREDENTIAL_VAR.set(resolution.secret)

            # ── Remote execution dispatch (executor=) ─────────────────────
            # When an executor is configured we package the call as an
            # envelope and hand it off; permission checks, lifecycle
            # events and result normalisation continue to run here so
            # remote and local tools look identical to the agent loop.
            if self.executor is not None and not args:
                from .executors.abstract import build_envelope_from_tool

                envelope = build_envelope_from_tool(
                    self,
                    arguments=resolved_kwargs,
                    permission_context=pctx,
                    trace_context=tool_tc,
                    timeout_seconds=self.remote_timeout_seconds,
                    webhook_callback_url=self.webhook_callback_url,
                )
                raw_result = await self.executor.execute(envelope)
            elif hasattr(validated_args, 'model_dump'):
                raw_result = await self._execute(*args, **resolved_kwargs)
            else:
                raw_result = await self._execute(*args, **kwargs)
        finally:
            # Always reset the ContextVar so the credential never leaks
            # to other concurrent tasks sharing the same context.
            _CREDENTIAL_VAR.reset(_cred_token)

        # Normalise to ToolResult
        if isinstance(raw_result, ToolResult):
            tool_result = raw_result
        elif isinstance(raw_result, dict) and 'status' in raw_result and 'result' in raw_result:
            try:
                tool_result = ToolResult(**raw_result)
            except Exception as e:
                self.logger.error("Error creating ToolResult from dict: %s", e)
                tool_result = ToolResult(
                    status="done_with_errors",
                    result=raw_result.get('result', []),
                    error=f"Error creating ToolResult: {e}",
                    metadata=raw_result.get('metadata', {})
                )
        elif raw_result is None:
            raise ValueError(
                "Tool execution returned None, expected a result."
            )
        else:
            self.logger.info("Tool %s executed successfully", self.name)
            tool_result = ToolResult(
                status="success",
                result=raw_result,
                metadata={
                    "tool_name": self.name,
                    "execution_time": datetime.now().isoformat()
                }
            )

        # ── FEAT-252 (TASK-1612): single-seam OutputScrubber hook ─────────
        # Scrub tool_result.result (and error/metadata) ONCE before return
        # so every tool — including python_repl — inherits secret redaction.
        # This is the ONLY place scrubbing happens on the way out; all
        # downstream callers receive a pre-scrubbed ToolResult.
        try:
            _scrubber = _default_scrubber()
            if tool_result.result is not None:
                tool_result = tool_result.model_copy(
                    update={"result": _scrubber.scrub(
                        tool_result.result, tool_name=_tool_name
                    )}
                )
            if tool_result.error is not None:
                tool_result = tool_result.model_copy(
                    update={"error": _scrubber.scrub(
                        tool_result.error, tool_name=_tool_name
                    )}
                )
            if tool_result.metadata:
                tool_result = tool_result.model_copy(
                    update={"metadata": _scrubber.scrub(
                        tool_result.metadata, tool_name=_tool_name
                    )}
                )
        except Exception as _scrub_exc:  # never let scrub errors break tool execution
            self.logger.warning(
                "OutputScrubber error in tool %s (non-fatal): %s",
                _tool_name, _scrub_exc,
            )

        # ── FEAT-176: emit AfterToolCallEvent on success ──────────────────
        _lc_dur = (time.perf_counter() - _lc_t0) * 1000
        await self.events.emit(AfterToolCallEvent(
            trace_context=tool_tc,
            tool_name=_tool_name,
            duration_ms=_lc_dur,
            result_status=tool_result.status,
            result_size_bytes=self._result_size(tool_result),
            source_type="tool",
            source_name=_tool_name,
        ))
        return tool_result

    except Exception as e:
        # Let ``AuthorizationRequired`` bubble up to ``ToolManager`` so it
        # can be converted into a structured ``authorization_required``
        # ToolResult (FEAT-107, TASK-748).  Imported lazily to avoid a
        # circular import with ``parrot.auth``.
        from ..auth.exceptions import AuthorizationRequired
        if isinstance(e, AuthorizationRequired):
            raise

        # FEAT-264: Let CredentialRequired bubble up to the surface so it
        # can render the appropriate UX (A2A suspend+link, Adaptive Card,
        # CLI URL).  Imported lazily to match the pattern above.
        from ..auth.credentials import CredentialRequired as _CR
        if isinstance(e, _CR):
            raise

        # ── FEAT-176: emit ToolCallFailedEvent before returning error ─────
        _lc_dur = (time.perf_counter() - _lc_t0) * 1000
        await self.events.emit(ToolCallFailedEvent(
            trace_context=tool_tc,
            tool_name=_tool_name,
            duration_ms=_lc_dur,
            error_type=type(e).__name__,
            error_message=str(e),
            source_type="tool",
            source_name=_tool_name,
        ))

        error_msg = f"Error in {self.name}: {str(e)}"
        self.logger.error("%s", error_msg)
        self.logger.debug("%s", traceback.format_exc())

        try:
            _scrubber = _default_scrubber()
            error_msg = _scrubber.scrub(error_msg, tool_name=_tool_name)
        except Exception:
            pass

        return ToolResult(
            status="error",
            result=None,
            error=error_msg,
            metadata={
                "tool_name": self.name,
                "error_type": type(e).__name__
            }
        )
    finally:
        # Always clear the per-call context so stale references don't linger.
        self._current_pctx = None

run async

run(*args, **kwargs) -> Any

Public alias for executing the tool directly without the ToolResult wrapper. Provides a direct way to get raw results instead of calling _execute().

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

async def run(self, *args, **kwargs) -> Any:
    """
    Public alias for executing the tool directly without the ToolResult wrapper.
    Provides a direct way to get raw results instead of calling _execute().
    """
    return await self._execute(*args, **kwargs)

to_static_url

to_static_url(file_path: Union[str, Path]) -> str

Convert an absolute file path to a static URL.

PARAMETER	DESCRIPTION
file_path	Absolute path to the file TYPE: Union[str, Path]

RETURNS	DESCRIPTION
str	URL-based path for serving the static file

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def to_static_url(self, file_path: Union[str, Path]) -> str:
    """
    Convert an absolute file path to a static URL.

    Args:
        file_path: Absolute path to the file

    Returns:
        URL-based path for serving the static file
    """
    if not self.static_dir:
        return str(file_path)

    file_path = Path(file_path)

    try:
        relative_path = file_path.relative_to(self.static_dir)
        return f"{self.static_url.rstrip('/')}/{relative_path}"
    except ValueError:
        self.logger.warning(
            "File %s is not within static directory %s", file_path, self.static_dir
        )
        return str(file_path)

relative_url

relative_url(url: str) -> str

Convert an absolute URL to a relative URL based on the base URL.

PARAMETER	DESCRIPTION
url	Absolute URL to convert TYPE: str

RETURNS	DESCRIPTION
str	Relative URL based on the base URL

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def relative_url(self, url: str) -> str:
    """
    Convert an absolute URL to a relative URL based on the base URL.

    Args:
        url: Absolute URL to convert

    Returns:
        Relative URL based on the base URL
    """
    parts = urlparse(url)
    if not parts.scheme or not parts.netloc:
        return url

    if (parts.scheme, parts.netloc) == self._base_scheme_netloc:
        return urlunparse((
            "", "", parts.path, parts.params, parts.query, parts.fragment
        ))
    return url

generate_filename

generate_filename(prefix: str = 'output', extension: str = '', include_timestamp: bool = True) -> str

Generate a unique filename with optional timestamp.

PARAMETER	DESCRIPTION
prefix	File prefix TYPE: str DEFAULT: 'output'
extension	File extension (with or without dot) TYPE: str DEFAULT: ''
include_timestamp	Whether to include timestamp TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	Generated filename

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def generate_filename(
    self,
    prefix: str = "output",
    extension: str = "",
    include_timestamp: bool = True
) -> str:
    """
    Generate a unique filename with optional timestamp.

    Args:
        prefix: File prefix
        extension: File extension (with or without dot)
        include_timestamp: Whether to include timestamp

    Returns:
        Generated filename
    """
    if include_timestamp:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{prefix}_{timestamp}"
    else:
        filename = prefix

    if extension:
        if not extension.startswith('.'):
            extension = f".{extension}"
        filename += extension

    return filename

validate_output_path

validate_output_path(file_path: Union[str, Path]) -> Path

Validate and ensure the output path is within allowed directories.

PARAMETER	DESCRIPTION
file_path	Path to validate TYPE: Union[str, Path]

RETURNS	DESCRIPTION
Path	Validated Path object

RAISES	DESCRIPTION
ValueError	If path is outside allowed directories

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def validate_output_path(self, file_path: Union[str, Path]) -> Path:
    """
    Validate and ensure the output path is within allowed directories.

    Args:
        file_path: Path to validate

    Returns:
        Validated Path object

    Raises:
        ValueError: If path is outside allowed directories
    """
    if not self.static_dir:
        return Path(file_path).resolve()

    file_path = Path(file_path).resolve()

    try:
        file_path.relative_to(self.static_dir.resolve())
    except ValueError as e:
        raise ValueError(
            f"Output path {file_path} must be within static directory {self.static_dir}"
        ) from e

    return file_path

ToolResult

Bases: BaseModel

Standardized tool result format.

spoken_content property

spoken_content: str

Returns content for voice synthesis.

has_display_content property

has_display_content: bool

Check if there's visual content to display.

AbstractToolkit

AbstractToolkit(**kwargs)

Bases: ABC

Abstract base class for creating toolkits - collections of related tools.

A toolkit automatically converts all public async methods into tools. Each method becomes a tool with: - Name: method name - Description: method docstring - Schema: automatically generated from type hints

Usage

class MyToolkit(AbstractToolkit): async def search_web(self, query: str) -> str: '''Search the web for information.''' # Implementation here return result

async def calculate(self, expression: str) -> float:
    '''Calculate a mathematical expression.'''
    # Implementation here
    return result

Get all tools

toolkit = MyToolkit() tools = toolkit.get_tools()

Initialize the toolkit.

PARAMETER	DESCRIPTION
executor	Optional AbstractToolExecutor instance. When provided, every ToolkitTool generated by this toolkit inherits it (and therefore runs off-process via the executor) unless the tool overrides it later.
webhook_callback_url	Forwarded to each generated tool's execute() envelope so async delivery can be enabled toolkit-wide.
remote_timeout_seconds	Forwarded as the per-tool remote timeout.
**kwargs	Additional configuration DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def __init__(self, **kwargs):
    """
    Initialize the toolkit.

    Args:
        executor: Optional ``AbstractToolExecutor`` instance. When
            provided, every ``ToolkitTool`` generated by this
            toolkit inherits it (and therefore runs off-process via
            the executor) unless the tool overrides it later.
        webhook_callback_url: Forwarded to each generated tool's
            ``execute()`` envelope so async delivery can be enabled
            toolkit-wide.
        remote_timeout_seconds: Forwarded as the per-tool remote
            timeout.
        **kwargs: Additional configuration
    """
    # Configuration
    self.return_direct = kwargs.get('return_direct', self.return_direct)
    self.base_url = kwargs.get('base_url', self.base_url)
    self.credential_provider = kwargs.get(
        'credential_provider', self.credential_provider
    )

    # Remote execution wiring — propagated to every generated tool.
    self.executor = kwargs.get('executor')
    self.webhook_callback_url = kwargs.get('webhook_callback_url')
    self.remote_timeout_seconds = int(
        kwargs.get('remote_timeout_seconds', 300)
    )

    # Capture init kwargs so build_envelope_from_tool can
    # reconstruct the toolkit on the remote side. The executor
    # instance itself is stripped by build_envelope_from_tool so
    # the remote side runs the tool in-process.
    self._init_kwargs = dict(kwargs)

    # Tool cache
    self._tool_cache: Dict[str, ToolkitTool] = {}
    self._tools_generated = False
    self.logger = logging.getLogger(self.__class__.__name__)

start async

start() -> None

Optional startup logic for the toolkit. Override in subclasses if needed.

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

async def start(self) -> None:
    """
    Optional startup logic for the toolkit.
    Override in subclasses if needed.
    """
    pass

stop async

stop() -> None

Optional shutdown logic for the toolkit. Override in subclasses if needed.

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

async def stop(self) -> None:
    """
    Optional shutdown logic for the toolkit.
    Override in subclasses if needed.
    """
    pass

cleanup async

cleanup() -> None

Optional cleanup logic for the toolkit. Override in subclasses if needed.

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

async def cleanup(self) -> None:
    """
    Optional cleanup logic for the toolkit.
    Override in subclasses if needed.
    """
    pass

get_tools

get_tools(permission_context: Optional[PermissionContext] = None, resolver: Optional[AbstractPermissionResolver] = None) -> List[AbstractTool]

Get all tools from this toolkit, optionally filtered by permissions.

Inspects all public methods and converts them to tools.

.. note::

Permission filtering requires ``await`` because resolvers are
async.  Use :meth:`get_tools_filtered` for the async path.
When *permission_context* and *resolver* are passed here the
method returns **all** tools (backward-compatible) — callers
must migrate to ``get_tools_filtered()`` for actual filtering.

PARAMETER	DESCRIPTION
permission_context	Ignored (kept for signature compat). Use :meth:get_tools_filtered instead. TYPE: Optional[PermissionContext] DEFAULT: None
resolver	Ignored (kept for signature compat). TYPE: Optional[AbstractPermissionResolver] DEFAULT: None

RETURNS	DESCRIPTION
List[AbstractTool]	List of AbstractTool instances (unfiltered).

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def get_tools(
    self,
    permission_context: Optional["PermissionContext"] = None,
    resolver: Optional["AbstractPermissionResolver"] = None,
) -> List[AbstractTool]:
    """
    Get all tools from this toolkit, optionally filtered by permissions.

    Inspects all public methods and converts them to tools.

    .. note::

        Permission filtering requires ``await`` because resolvers are
        async.  Use :meth:`get_tools_filtered` for the async path.
        When *permission_context* and *resolver* are passed here the
        method returns **all** tools (backward-compatible) — callers
        must migrate to ``get_tools_filtered()`` for actual filtering.

    Args:
        permission_context: Ignored (kept for signature compat).
            Use :meth:`get_tools_filtered` instead.
        resolver: Ignored (kept for signature compat).

    Returns:
        List of AbstractTool instances (unfiltered).
    """
    if not self._tools_generated or not self._tool_cache:
        # Generate tools if not yet done
        self._generate_tools()

    return list(self._tool_cache.values())

get_tools_filtered async

get_tools_filtered(permission_context: PermissionContext, resolver: AbstractPermissionResolver) -> List[AbstractTool]

Get tools filtered by async permission resolver.

This is the async-aware version of :meth:get_tools that supports Layer 1 preventive filtering through the resolver.

PARAMETER	DESCRIPTION
permission_context	User context for permission filtering. TYPE: PermissionContext
resolver	Permission resolver for checking access (async). TYPE: AbstractPermissionResolver

RETURNS	DESCRIPTION
List[AbstractTool]	Filtered list of AbstractTool instances the user may execute.

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

async def get_tools_filtered(
    self,
    permission_context: "PermissionContext",
    resolver: "AbstractPermissionResolver",
) -> List[AbstractTool]:
    """Get tools filtered by async permission resolver.

    This is the async-aware version of :meth:`get_tools` that supports
    Layer 1 preventive filtering through the resolver.

    Args:
        permission_context: User context for permission filtering.
        resolver: Permission resolver for checking access (async).

    Returns:
        Filtered list of AbstractTool instances the user may execute.
    """
    all_tools = self.get_tools()
    return await resolver.filter_tools(permission_context, all_tools)

get_tools_sync

get_tools_sync(permission_context: Optional[PermissionContext] = None, resolver: Optional[AbstractPermissionResolver] = None) -> List[AbstractTool]

Synchronous alias for get_tools(). Returns all tools (unfiltered).

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def get_tools_sync(
    self,
    permission_context: Optional["PermissionContext"] = None,
    resolver: Optional["AbstractPermissionResolver"] = None,
) -> List[AbstractTool]:
    """Synchronous alias for get_tools(). Returns all tools (unfiltered)."""
    return self.get_tools()

get_tool

get_tool(name: str) -> Optional[AbstractTool]

Get a specific tool by name.

PARAMETER	DESCRIPTION
name	Tool name TYPE: str

RETURNS	DESCRIPTION
Optional[AbstractTool]	AbstractTool instance or None if not found

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def get_tool(self, name: str) -> Optional[AbstractTool]:
    """
    Get a specific tool by name.

    Args:
        name: Tool name

    Returns:
        AbstractTool instance or None if not found
    """
    if not self._tools_generated:
        self._generate_tools()  # Ensure tools are generated

    return self._tool_cache.get(name)

list_tool_names

list_tool_names() -> List[str]

Get a list of all tool names in this toolkit.

RETURNS	DESCRIPTION
List[str]	List of tool names

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def list_tool_names(self) -> List[str]:
    """
    Get a list of all tool names in this toolkit.

    Returns:
        List of tool names
    """
    if not self._tools_generated:
        self._generate_tools()

    return list(self._tool_cache.keys())

get_toolkit_info

get_toolkit_info() -> Dict[str, Any]

Get information about this toolkit.

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary with toolkit information

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def get_toolkit_info(self) -> Dict[str, Any]:
    """
    Get information about this toolkit.

    Returns:
        Dictionary with toolkit information
    """
    if not self._tools_generated:
        self._generate_tools()

    tools = list(self._tool_cache.values())

    return {
        "toolkit_name": self.__class__.__name__,
        "tool_count": len(tools),
        "tool_names": [tool.name for tool in tools],
        "tool_descriptions": {tool.name: tool.description for tool in tools},
        "return_direct": self.return_direct,
        "base_url": self.base_url
    }

ToolkitTool

ToolkitTool(name: str, bound_method: callable, description: str = None, args_schema: Type[BaseModel] = None, **kwargs)

Bases: AbstractTool

A specialized AbstractTool that wraps a method from a toolkit.

Initialize a toolkit tool.

PARAMETER	DESCRIPTION
name	Tool name TYPE: str
bound_method	The bound coroutine method to wrap TYPE: callable
description	Tool description TYPE: str DEFAULT: None
args_schema	Pydantic model for arguments TYPE: Type[BaseModel] DEFAULT: None
**kwargs	Additional arguments DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/tools/toolkit.py

def __init__(
    self,
    name: str,
    bound_method: callable,
    description: str = None,
    args_schema: Type[BaseModel] = None,
    **kwargs
):
    """
    Initialize a toolkit tool.

    Args:
        name: Tool name
        bound_method: The bound coroutine method to wrap
        description: Tool description
        args_schema: Pydantic model for arguments
        **kwargs: Additional arguments
    """
    self.bound_method = bound_method

    # Set up the tool
    super().__init__(
        name=name,
        description=description or bound_method.__doc__ or f"Tool: {name}",
        **kwargs
    )

    # Set the args schema
    if args_schema:
        self.args_schema = args_schema
    else:
        # Try to generate schema from method signature
        self.args_schema = self._generate_args_schema_from_method()

ToolkitRegistry

Registry for supported toolkits with lazy loading.

.. deprecated:: Use ToolManager with discovery instead. This class is maintained for backward compatibility.

get_registry classmethod

get_registry() -> Dict[str, Type[AbstractToolkit]]

Get the toolkit registry, initializing lazily if needed.

Source code in packages/ai-parrot/src/parrot/tools/registry.py

@classmethod
def get_registry(cls) -> Dict[str, Type["AbstractToolkit"]]:
    """Get the toolkit registry, initializing lazily if needed."""
    if cls._registry is None:
        cls._registry = _discover_toolkits()
    return cls._registry

get classmethod

get(name: str) -> Type[AbstractToolkit]

Get a toolkit class by name.

Source code in packages/ai-parrot/src/parrot/tools/registry.py

@classmethod
def get(cls, name: str) -> Type["AbstractToolkit"]:
    """Get a toolkit class by name."""
    registry = cls.get_registry()
    return registry.get(name.lower())

list_toolkits classmethod

list_toolkits() -> list

List all available toolkit names.

Source code in packages/ai-parrot/src/parrot/tools/registry.py

@classmethod
def list_toolkits(cls) -> list:
    """List all available toolkit names."""
    return list(cls.get_registry().keys())

register classmethod

register(name: str, toolkit_class: Type[AbstractToolkit]) -> None

Register a custom toolkit.

Source code in packages/ai-parrot/src/parrot/tools/registry.py

@classmethod
def register(cls, name: str, toolkit_class: Type["AbstractToolkit"]) -> None:
    """Register a custom toolkit."""
    registry = cls.get_registry()
    registry[name.lower()] = toolkit_class

AbstractToolExecutor

Bases: ABC

Pluggable transport that runs a tool somewhere other than here.

Concrete executors translate a :class:ToolExecutionEnvelope into whatever protocol the remote runtime speaks (HTTP, gRPC, k8s API, Redis Streams, etc.) and return a :class:ToolResult once the remote side finishes — either by waiting synchronously up to envelope.timeout_seconds or by returning a pending ToolResult and arranging for the final result to arrive via webhook.

Concrete implementations:

• :class:LocalToolExecutor — in-process; reference / tests
• :class:K8sToolExecutor — ephemeral Pod via kubernetes-asyncio
• :class:QworkerToolExecutor — Qworker service (HTTP / Redis)

execute abstractmethod async

execute(envelope: ToolExecutionEnvelope) -> 'ToolResult'

Run the tool described by envelope and return its ToolResult.

Implementations must:

• raise :class:asyncio.TimeoutError (or a structured ToolResult(status="error", error="timeout")) when the remote side fails to respond within envelope.timeout_seconds.
• never re-run permission checks. Those happen on the caller.
• propagate envelope.trace_context to the worker so the trace stays connected.

RETURNS	DESCRIPTION
'ToolResult'	The final ToolResult, or a ToolResult(status="pending")
'ToolResult'	with metadata["job_id"] set when the executor was
'ToolResult'	constructed with a webhook delivery configuration.

Source code in packages/ai-parrot/src/parrot/tools/executors/abstract.py

@abstractmethod
async def execute(
    self, envelope: ToolExecutionEnvelope
) -> "ToolResult":
    """Run the tool described by *envelope* and return its ToolResult.

    Implementations must:

    * raise :class:`asyncio.TimeoutError` (or a structured
      ``ToolResult(status="error", error="timeout")``) when the
      remote side fails to respond within ``envelope.timeout_seconds``.
    * never re-run permission checks. Those happen on the caller.
    * propagate ``envelope.trace_context`` to the worker so the trace
      stays connected.

    Returns:
        The final ToolResult, or a ``ToolResult(status="pending")``
        with ``metadata["job_id"]`` set when the executor was
        constructed with a webhook delivery configuration.
    """

close abstractmethod async

close() -> None

Release any pooled resources (HTTP sessions, k8s clients, etc.).

Idempotent: calling close() multiple times must not raise.

Source code in packages/ai-parrot/src/parrot/tools/executors/abstract.py

@abstractmethod
async def close(self) -> None:
    """Release any pooled resources (HTTP sessions, k8s clients, etc.).

    Idempotent: calling ``close()`` multiple times must not raise.
    """

LocalToolExecutor

Bases: AbstractToolExecutor

Executor that runs the tool in the current Python process.

Used as the reference implementation: it imports the tool by path, instantiates it from envelope.tool_init_kwargs, and awaits its _execute(**envelope.arguments). Because it shares the runner module with the worker entrypoint, this is what the K8sToolExecutor worker pod ends up doing inside its own process — and what unit tests can exercise without ceremony.

ToolExecutionEnvelope

Bases: BaseModel

The wire-format payload describing a single remote tool invocation.

ATTRIBUTE	DESCRIPTION
tool_import_path	Dotted Python path of the tool class, formatted as "<module>:<qualname>" so the remote worker can do importlib.import_module(module) and getattr(cls). TYPE: str
tool_init_kwargs	Constructor arguments captured from the caller's instance. Forwarded as cls(**tool_init_kwargs) on the remote side. The executor kwarg is stripped before transit so the remote tool runs locally. TYPE: Dict[str, Any]
method_name	For ToolkitTool envelopes, the name of the bound method to invoke on the reconstructed toolkit. None for plain AbstractTool subclasses. TYPE: Optional[str]
arguments	Validated tool arguments (the kwargs that would normally be passed to _execute). TYPE: Dict[str, Any]
permission_context	JSON projection of the caller's PermissionContext. The remote side does NOT re-run permission checks — Layer 2 enforcement happens on the caller before the envelope is sent. This is informational. TYPE: Optional[Dict[str, Any]]
trace_context	JSON projection of the parent span so the remote runtime can mint a child span and keep the trace connected. TYPE: Optional[Dict[str, Any]]
timeout_seconds	Maximum wall-clock seconds to wait for the remote runtime to return a result. TYPE: int
webhook_callback_url	When set, the executor returns immediately with a "pending" ToolResult; the remote runtime POSTs the final ToolResult to this URL when it completes. The webhook handler is registered separately. TYPE: Optional[str]
envelope_version	Schema version. Bumped when the contract changes in a backwards-incompatible way. TYPE: int

MCPToolManagerMixin

Mixin to add MCP capabilities to ToolManager.

This mixin adds the following capabilities: - Connect to MCP servers (HTTP, SSE, WebSocket, stdio, QUIC) - Register MCP tools as proxy tools in the ToolManager - Generate OpenAI-compatible MCP definitions - Manage server lifecycle (connect, disconnect, reconfigure)

ATTRIBUTE	DESCRIPTION
_mcp_clients	Dictionary mapping server names to MCPClient instances
_mcp_configs	Dictionary mapping server names to MCPServerConfig
_mcp_logger	Logger for MCP operations

add_mcp_server async

add_mcp_server(config: 'MCPServerConfig', context: Optional['ReadonlyContext'] = None) -> List[str]

Add MCP server with context-aware tool registration.

Connects to an MCP server and registers its tools as proxy tools in this ToolManager. The tools are filtered based on the config's allowed_tools, blocked_tools, and tool_filter settings.

PARAMETER	DESCRIPTION
config	MCPServerConfig with connection and filtering options TYPE: 'MCPServerConfig'
context	Optional ReadonlyContext for dynamic filtering decisions TYPE: Optional['ReadonlyContext'] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	List of registered tool names (prefixed with mcp_{server_name}_)

RAISES	DESCRIPTION
MCPConnectionError	If connection to server fails

Example

config = MCPServerConfig(
    name="web-tools",
    url="https://mcp.example.com/sse",
    transport="sse",
    allowed_tools=["search", "fetch"]
)
tools = await tool_manager.add_mcp_server(config)
# tools = ['mcp_web-tools_search', 'mcp_web-tools_fetch']

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def add_mcp_server(
    self,
    config: 'MCPServerConfig',
    context: Optional['ReadonlyContext'] = None
) -> List[str]:
    """Add MCP server with context-aware tool registration.

    Connects to an MCP server and registers its tools as proxy tools
    in this ToolManager. The tools are filtered based on the config's
    allowed_tools, blocked_tools, and tool_filter settings.

    Args:
        config: MCPServerConfig with connection and filtering options
        context: Optional ReadonlyContext for dynamic filtering decisions

    Returns:
        List of registered tool names (prefixed with mcp_{server_name}_)

    Raises:
        MCPConnectionError: If connection to server fails

    Example:
        ```python
        config = MCPServerConfig(
            name="web-tools",
            url="https://mcp.example.com/sse",
            transport="sse",
            allowed_tools=["search", "fetch"]
        )
        tools = await tool_manager.add_mcp_server(config)
        # tools = ['mcp_web-tools_search', 'mcp_web-tools_fetch']
        ```
    """
    from ..mcp.integration import MCPClient, MCPToolProxy

    # Acquire the instance lock to guard concurrent add/remove on the same
    # ToolManager (e.g. two simultaneous POST /mcp-servers for the same agent).
    async with self._mcp_lock:
        # Check if server with same name already exists
        if config.name in self._mcp_clients:
            self._mcp_logger.warning(
                "MCP server '%s' already exists. "
                "Use reconfigure_mcp_server() to update it.",
                config.name,
            )
            return []

        client = MCPClient(config)

        try:
            await client.connect()
            self._mcp_clients[config.name] = client
            self._mcp_configs[config.name] = config

            available_tools = await client.get_available_tools()
            registered_tools = []

            # Resolve optional private filtering helpers on MCPClient.
            # These are internal methods that provide predicate-based tool
            # selection.  We access them via getattr so that if MCPClient's
            # internal API changes, we degrade gracefully to basic filtering
            # rather than crashing.
            _create_temp = getattr(client, '_create_temp_tool_for_filtering', None)
            _is_selected = getattr(client, '_is_tool_selected', None)

            for tool_def in available_tools:
                tool_name = tool_def.get('name', 'unknown')

                # Basic allowed/blocked list filtering
                if self._should_skip_mcp_tool(tool_name, config):
                    continue

                # Dynamic predicate filtering (if supported by MCPClient)
                if _create_temp is not None and _is_selected is not None:
                    if not _is_selected(_create_temp(tool_def), context):
                        self._mcp_logger.debug(
                            "Tool %s filtered out by predicate", tool_name
                        )
                        continue

                # Create proxy tool
                proxy_tool = MCPToolProxy(
                    mcp_tool_def=tool_def,
                    mcp_client=client,
                    server_name=config.name,
                    require_confirmation=getattr(config, 'require_confirmation', False),
                )

                # FEAT-264: When the MCP server is configured with
                # inject_broker_credential=True, tag each proxy tool with the
                # server name as its credential_provider so the tool-loop seam
                # (AbstractTool.execute) resolves per-user credentials via the
                # broker before calling _execute().  The broker-resolved token
                # then flows to MCPClientConfig.get_headers() via the ContextVar.
                if getattr(config, 'inject_broker_credential', False):
                    proxy_tool.credential_provider = config.name

                # Register in self (ToolManager)
                self.register_tool(proxy_tool)
                registered_tools.append(proxy_tool.name)
                self._mcp_logger.info("Registered MCP tool: %s", proxy_tool.name)

            transport_type = (
                config.transport if config.transport != "auto" else "detected"
            )
            self._mcp_logger.info(
                "Successfully added MCP server %s (%s transport) with %d tools",
                config.name,
                transport_type,
                len(registered_tools),
            )
            return registered_tools

        except Exception as e:
            self._mcp_logger.error("Failed to add MCP server %s: %s", config.name, e)
            await self._cleanup_failed_mcp_client(config.name, client)
            raise

add_database_mcp async

add_database_mcp(name: str, project_id: str, toolbox_path: str = './toolbox', database_type: str = 'bigquery', context: Optional['ReadonlyContext'] = None, extra_env: Optional[Dict[str, str]] = None) -> List[str]

Add a database MCP server using the genai-toolbox.

PARAMETER	DESCRIPTION
name	Name for the MCP server instance TYPE: str
project_id	GCP Project ID to use TYPE: str
toolbox_path	Path to the toolbox binary (default: ./toolbox) TYPE: str DEFAULT: './toolbox'
database_type	Type of database (default: bigquery) TYPE: str DEFAULT: 'bigquery'
context	Optional context for filtering TYPE: Optional['ReadonlyContext'] DEFAULT: None
extra_env	Optional extra environment variables TYPE: Optional[Dict[str, str]] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def add_database_mcp(
    self,
    name: str,
    project_id: str,
    toolbox_path: str = "./toolbox",
    database_type: str = "bigquery",
    context: Optional['ReadonlyContext'] = None,
    extra_env: Optional[Dict[str, str]] = None
) -> List[str]:
    """Add a database MCP server using the genai-toolbox.

    Args:
        name: Name for the MCP server instance
        project_id: GCP Project ID to use
        toolbox_path: Path to the toolbox binary (default: ./toolbox)
        database_type: Type of database (default: bigquery)
        context: Optional context for filtering
        extra_env: Optional extra environment variables

    Returns:
        List of registered tool names
    """
    from ..mcp.client import MCPClientConfig as MCPServerConfig

    env = {"BIGQUERY_PROJECT": project_id}
    if extra_env:
        env.update(extra_env)

    config = MCPServerConfig(
        name=name,
        transport="stdio",
        command=toolbox_path,
        args=["--prebuilt", database_type, "--stdio"],
        env=env
    )

    return await self.add_mcp_server(config, context)

add_github_mcp async

add_github_mcp(name: str = 'github', personal_access_token: Optional[str] = None, context: Optional['ReadonlyContext'] = None, **kwargs) -> List[str]

Add GitHub MCP server using npx.

PARAMETER	DESCRIPTION
name	Server name (default: "github") TYPE: str DEFAULT: 'github'
personal_access_token	GitHub PAT. If None, checks GITHUB_PERSONAL_ACCESS_TOKEN env var. TYPE: Optional[str] DEFAULT: None
context	Optional context for filtering TYPE: Optional['ReadonlyContext'] DEFAULT: None
**kwargs	Additional config arguments DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

RAISES	DESCRIPTION
ValueError	If token is missing

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def add_github_mcp(
    self,
    name: str = "github",
    personal_access_token: Optional[str] = None,
    context: Optional['ReadonlyContext'] = None,
    **kwargs
) -> List[str]:
    """Add GitHub MCP server using npx.

    Args:
        name: Server name (default: "github")
        personal_access_token: GitHub PAT. If None, checks GITHUB_PERSONAL_ACCESS_TOKEN env var.
        context: Optional context for filtering
        **kwargs: Additional config arguments

    Returns:
        List of registered tool names

    Raises:
        ValueError: If token is missing
    """
    import os
    from ..mcp.client import MCPClientConfig as MCPServerConfig

    token = personal_access_token or os.environ.get("GITHUB_PERSONAL_ACCESS_TOKEN")
    if not token:
        raise ValueError(
            "GitHub Personal Access Token is required. "
            "Pass it as argument or set GITHUB_PERSONAL_ACCESS_TOKEN env var."
        )

    env = os.environ.copy()
    env["GITHUB_PERSONAL_ACCESS_TOKEN"] = token

    config = MCPServerConfig(
        name=name,
        transport="stdio",
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env=env,
        **kwargs
    )

    return await self.add_mcp_server(config, context)

add_github_remote_mcp async

add_github_remote_mcp(name: str = 'github-remote', personal_access_token: Optional[str] = None, toolsets: Union[List[str], str] = 'repos,issues', readonly: bool = True, lockdown: bool = False, context: Optional['ReadonlyContext'] = None, **kwargs) -> List[str]

Add GitHub Remote MCP server (insiders) via HTTP.

PARAMETER	DESCRIPTION
name	Server name (default: "github-remote") TYPE: str DEFAULT: 'github-remote'
personal_access_token	GitHub PAT. If None, checks GITHUB_PERSONAL_ACCESS_TOKEN. TYPE: Optional[str] DEFAULT: None
toolsets	List or comma-separated string of toolsets (default: "repos,issues") TYPE: Union[List[str], str] DEFAULT: 'repos,issues'
readonly	Whether to run in readonly mode (default: True) TYPE: bool DEFAULT: True
lockdown	Whether to run in lockdown mode (default: False) TYPE: bool DEFAULT: False
context	Optional context for filtering TYPE: Optional['ReadonlyContext'] DEFAULT: None
**kwargs	Additional config arguments DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

RAISES	DESCRIPTION
ValueError	If token is missing

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def add_github_remote_mcp(
    self,
    name: str = "github-remote",
    personal_access_token: Optional[str] = None,
    toolsets: Union[List[str], str] = "repos,issues",
    readonly: bool = True,
    lockdown: bool = False,
    context: Optional['ReadonlyContext'] = None,
    **kwargs
) -> List[str]:
    """Add GitHub Remote MCP server (insiders) via HTTP.

    Args:
        name: Server name (default: "github-remote")
        personal_access_token: GitHub PAT. If None, checks GITHUB_PERSONAL_ACCESS_TOKEN.
        toolsets: List or comma-separated string of toolsets (default: "repos,issues")
        readonly: Whether to run in readonly mode (default: True)
        lockdown: Whether to run in lockdown mode (default: False)
        context: Optional context for filtering
        **kwargs: Additional config arguments

    Returns:
        List of registered tool names

    Raises:
        ValueError: If token is missing
    """
    import os
    from ..mcp.client import MCPClientConfig as MCPServerConfig

    token = personal_access_token or os.environ.get("GITHUB_PERSONAL_ACCESS_TOKEN")
    if not token:
        raise ValueError(
            "GitHub Personal Access Token is required. "
            "Pass it as argument or set GITHUB_PERSONAL_ACCESS_TOKEN env var."
        )

    # Format toolsets
    if isinstance(toolsets, list):
        toolsets_str = ",".join(toolsets)
    else:
        toolsets_str = toolsets

    headers = {
        "X-MCP-Toolsets": toolsets_str,
        "X-MCP-Readonly": str(readonly).lower(),
        "X-MCP-Lockdown": str(lockdown).lower(),
        "Authorization": f"token {token}",
        "User-Agent": "ai-parrot-mcp-client"
    }

    # Merge with any extra headers passed in kwargs
    if "headers" in kwargs:
        headers.update(kwargs.pop("headers"))

    config = MCPServerConfig(
        name=name,
        transport="http",
        url="https://api.githubcopilot.com/mcp/",
        headers=headers,
        **kwargs
    )

    return await self.add_mcp_server(config, context)

remove_mcp_server async

remove_mcp_server(server_name: str) -> bool

Remove an MCP server and unregister its tools.

Disconnects from the specified MCP server and removes all tools that were registered from it.

PARAMETER	DESCRIPTION
server_name	Name of the MCP server to remove TYPE: str

RETURNS	DESCRIPTION
bool	True if server was removed, False if not found

Example

removed = await tool_manager.remove_mcp_server("web-tools")
if removed:
    print("Server removed successfully")

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def remove_mcp_server(self, server_name: str) -> bool:
    """Remove an MCP server and unregister its tools.

    Disconnects from the specified MCP server and removes all tools
    that were registered from it.

    Args:
        server_name: Name of the MCP server to remove

    Returns:
        True if server was removed, False if not found

    Example:
        ```python
        removed = await tool_manager.remove_mcp_server("web-tools")
        if removed:
            print("Server removed successfully")
        ```
    """
    async with self._mcp_lock:
        if server_name not in self._mcp_clients:
            self._mcp_logger.warning("MCP server %s not found", server_name)
            return False

        client = self._mcp_clients[server_name]

        # Find and remove all tools from this server.
        # Iterate over a copy of keys since we're modifying the dict.
        tools_to_remove = [
            name for name, tool in list(self._tools.items())
            if hasattr(tool, 'server_name') and tool.server_name == server_name
        ]

        for tool_name in tools_to_remove:
            self.unregister_tool(tool_name)
            self._mcp_logger.debug("Unregistered MCP tool: %s", tool_name)

        # Disconnect client
        try:
            await client.disconnect()
        except Exception as e:
            self._mcp_logger.warning(
                "Error disconnecting from %s: %s", server_name, e
            )

        # Clean up registries
        self._mcp_clients.pop(server_name, None)
        self._mcp_configs.pop(server_name, None)

    self._mcp_logger.info(
        "Removed MCP server %s and %d tools", server_name, len(tools_to_remove)
    )
    return True

reconfigure_mcp_server async

reconfigure_mcp_server(config: 'MCPServerConfig', context: Optional['ReadonlyContext'] = None) -> List[str]

Reconfigure an existing MCP server with new settings.

This removes the existing server and re-adds it with new configuration. Useful for changing auth tokens, allowed tools, etc.

PARAMETER	DESCRIPTION
config	New MCPServerConfig (name must match existing server) TYPE: 'MCPServerConfig'
context	Optional ReadonlyContext for filtering TYPE: Optional['ReadonlyContext'] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	List of newly registered tool names

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def reconfigure_mcp_server(
    self,
    config: 'MCPServerConfig',
    context: Optional['ReadonlyContext'] = None
) -> List[str]:
    """Reconfigure an existing MCP server with new settings.

    This removes the existing server and re-adds it with new configuration.
    Useful for changing auth tokens, allowed tools, etc.

    Args:
        config: New MCPServerConfig (name must match existing server)
        context: Optional ReadonlyContext for filtering

    Returns:
        List of newly registered tool names
    """
    await self.remove_mcp_server(config.name)
    return await self.add_mcp_server(config, context)

disconnect_all_mcp async

disconnect_all_mcp()

Disconnect from all MCP servers and unregister their tools.

Cleanly disconnects all MCP server connections and removes all associated proxy tools from this ToolManager. Safe to call at session teardown or during a full reset.

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

async def disconnect_all_mcp(self):
    """Disconnect from all MCP servers and unregister their tools.

    Cleanly disconnects all MCP server connections and removes all
    associated proxy tools from this ToolManager.  Safe to call at
    session teardown or during a full reset.
    """
    # Snapshot names first — remove_mcp_server modifies self._mcp_clients
    server_names = list(self._mcp_clients.keys())
    for name in server_names:
        try:
            await self.remove_mcp_server(name)
            self._mcp_logger.debug("Disconnected from %s", name)
        except Exception as e:
            self._mcp_logger.warning("Error disconnecting from %s: %s", name, e)

list_mcp_servers

list_mcp_servers() -> List[str]

List all connected MCP server names.

RETURNS	DESCRIPTION
List[str]	List of server names that are currently connected

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def list_mcp_servers(self) -> List[str]:
    """List all connected MCP server names.

    Returns:
        List of server names that are currently connected
    """
    return list(self._mcp_clients.keys())

get_mcp_client

get_mcp_client(server_name: str) -> Optional['MCPClient']

Get MCP client by server name.

PARAMETER	DESCRIPTION
server_name	Name of the server TYPE: str

RETURNS	DESCRIPTION
Optional['MCPClient']	MCPClient instance or None if not found

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def get_mcp_client(self, server_name: str) -> Optional['MCPClient']:
    """Get MCP client by server name.

    Args:
        server_name: Name of the server

    Returns:
        MCPClient instance or None if not found
    """
    return self._mcp_clients.get(server_name)

get_mcp_config

get_mcp_config(server_name: str) -> Optional['MCPServerConfig']

Get MCP server configuration by name.

PARAMETER	DESCRIPTION
server_name	Name of the server TYPE: str

RETURNS	DESCRIPTION
Optional['MCPServerConfig']	MCPServerConfig or None if not found

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def get_mcp_config(self, server_name: str) -> Optional['MCPServerConfig']:
    """Get MCP server configuration by name.

    Args:
        server_name: Name of the server

    Returns:
        MCPServerConfig or None if not found
    """
    return self._mcp_configs.get(server_name)

get_openai_mcp_definitions

get_openai_mcp_definitions(server_names: Optional[List[str]] = None) -> List[Dict[str, Any]]

Get OpenAI-compatible MCP tool definitions.

Generates definitions in the format OpenAI accepts for native MCP server injection. Only servers with HTTP-based transports (http, sse, websocket) are included since OpenAI doesn't support stdio.

The returned format is:

{
    "type": "mcp",
    "server_label": "server_name",
    "server_description": "...",  # optional
    "server_url": "https://...",
    "require_approval": "never" | "always",
    "allowed_tools": [...],  # optional
    "headers": {...}  # optional
}

PARAMETER	DESCRIPTION
server_names	Optional list of server names to include. If None, includes all HTTP-based servers. TYPE: Optional[List[str]] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	List of OpenAI-compatible MCP definitions

Example

# Get all definitions
tools = tool_manager.get_openai_mcp_definitions()

# Get specific servers only
tools = tool_manager.get_openai_mcp_definitions(['web-tools'])

# Use with OpenAI
from openai import OpenAI
client = OpenAI()
resp = client.responses.create(
    model="gpt-5",
    tools=tools,
    input="Search for AI news"
)

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def get_openai_mcp_definitions(
    self,
    server_names: Optional[List[str]] = None
) -> List[Dict[str, Any]]:
    """Get OpenAI-compatible MCP tool definitions.

    Generates definitions in the format OpenAI accepts for native MCP
    server injection. Only servers with HTTP-based transports (http, sse,
    websocket) are included since OpenAI doesn't support stdio.

    The returned format is:
    ```python
    {
        "type": "mcp",
        "server_label": "server_name",
        "server_description": "...",  # optional
        "server_url": "https://...",
        "require_approval": "never" | "always",
        "allowed_tools": [...],  # optional
        "headers": {...}  # optional
    }
    ```

    Args:
        server_names: Optional list of server names to include.
                     If None, includes all HTTP-based servers.

    Returns:
        List of OpenAI-compatible MCP definitions

    Example:
        ```python
        # Get all definitions
        tools = tool_manager.get_openai_mcp_definitions()

        # Get specific servers only
        tools = tool_manager.get_openai_mcp_definitions(['web-tools'])

        # Use with OpenAI
        from openai import OpenAI
        client = OpenAI()
        resp = client.responses.create(
            model="gpt-5",
            tools=tools,
            input="Search for AI news"
        )
        ```
    """
    definitions = []

    servers_to_include = server_names or list(self._mcp_configs.keys())

    for name in servers_to_include:
        config = self._mcp_configs.get(name)
        if not config:
            self._mcp_logger.debug(f"Server {name} not found, skipping")
            continue

        # Only HTTP-based transports work with OpenAI
        transport = getattr(config, 'transport', 'auto')
        if transport not in ("http", "sse", "websocket"):
            self._mcp_logger.debug(
                f"Skipping {name}: transport '{transport}' not supported by OpenAI"
            )
            continue

        # Must have a URL
        url = getattr(config, 'url', None)
        if not url:
            self._mcp_logger.debug(f"Skipping {name}: no URL configured")
            continue

        # Build definition
        require_confirmation = getattr(config, 'require_confirmation', False)

        definition = {
            "type": "mcp",
            "server_label": name,
            "server_url": url,
            "require_approval": "always" if require_confirmation else "never",
        }

        # Add optional description
        description = getattr(config, 'description', None)
        if description:
            definition["server_description"] = description

        # Add optional allowed_tools
        allowed_tools = getattr(config, 'allowed_tools', None)
        if allowed_tools:
            definition["allowed_tools"] = allowed_tools

        # Add optional headers
        headers = getattr(config, 'headers', None)
        if headers:
            definition["headers"] = headers

        definitions.append(definition)

    return definitions

get_mcp_tools

get_mcp_tools(server_name: Optional[str] = None) -> List[Any]

Get all MCP tools, optionally filtered by server.

PARAMETER	DESCRIPTION
server_name	Optional server name to filter by. If None, returns all MCP tools. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[Any]	List of MCPToolProxy instances

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def get_mcp_tools(self, server_name: Optional[str] = None) -> List[Any]:
    """Get all MCP tools, optionally filtered by server.

    Args:
        server_name: Optional server name to filter by.
                    If None, returns all MCP tools.

    Returns:
        List of MCPToolProxy instances
    """
    from ..mcp.integration import MCPToolProxy

    tools = []
    for tool in self._tools.values():
        if isinstance(tool, MCPToolProxy):
            if server_name is None or tool.server_name == server_name:
                tools.append(tool)
    return tools

has_mcp_servers

has_mcp_servers() -> bool

Check if any MCP servers are connected.

RETURNS	DESCRIPTION
bool	True if at least one MCP server is connected

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def has_mcp_servers(self) -> bool:
    """Check if any MCP servers are connected.

    Returns:
        True if at least one MCP server is connected
    """
    return len(self._mcp_clients) > 0

get_mcp_server_info

get_mcp_server_info() -> Dict[str, Dict[str, Any]]

Get detailed information about all connected MCP servers.

RETURNS	DESCRIPTION
Dict[str, Dict[str, Any]]	Dictionary mapping server names to info dicts containing:
Dict[str, Dict[str, Any]]	transport: Transport type (http, sse, stdio, etc.)
Dict[str, Dict[str, Any]]	url: Server URL (if applicable)
Dict[str, Dict[str, Any]]	tool_count: Number of tools registered from this server
Dict[str, Dict[str, Any]]	connected: Whether client is connected

Source code in packages/ai-parrot/src/parrot/tools/mcp_mixin.py

def get_mcp_server_info(self) -> Dict[str, Dict[str, Any]]:
    """Get detailed information about all connected MCP servers.

    Returns:
        Dictionary mapping server names to info dicts containing:
        - transport: Transport type (http, sse, stdio, etc.)
        - url: Server URL (if applicable)
        - tool_count: Number of tools registered from this server
        - connected: Whether client is connected
    """
    info = {}

    for name, config in self._mcp_configs.items():
        client = self._mcp_clients.get(name)
        tools = self.get_mcp_tools(name)

        info[name] = {
            "transport": getattr(config, 'transport', 'unknown'),
            "url": getattr(config, 'url', None),
            "tool_count": len(tools),
            "connected": client is not None and getattr(client, '_connected', False),
            "description": getattr(config, 'description', None),
        }

    return info

ToJsonTool

ToJsonTool(name: Optional[str] = None, description: Optional[str] = None, output_dir: Optional[Union[str, Path]] = None, base_url: Optional[str] = None, static_dir: Optional[Union[str, Path]] = None, routing_meta: Optional[Dict] = None, executor: Optional[AbstractToolExecutor] = None, webhook_callback_url: Optional[str] = None, remote_timeout_seconds: int = 300, **kwargs)

Bases: AbstractTool

Tool to convert data to JSON using datamodel.parsers.json.

Source code in packages/ai-parrot/src/parrot/tools/abstract.py

def __init__(
    self,
    name: Optional[str] = None,
    description: Optional[str] = None,
    output_dir: Optional[Union[str, Path]] = None,
    base_url: Optional[str] = None,
    static_dir: Optional[Union[str, Path]] = None,
    routing_meta: Optional[Dict] = None,
    executor: Optional["AbstractToolExecutor"] = None,
    webhook_callback_url: Optional[str] = None,
    remote_timeout_seconds: int = 300,
    **kwargs
):
    """
    Initialize the tool.

    Args:
        name: Tool name (defaults to class name)
        description: Tool description
        output_dir: Directory for output files (if tool generates files)
        base_url: Base URL for serving static files
        static_dir: Static directory path
        routing_meta: Optional routing hints dict for CapabilityRegistry.
            Supported keys: ``"description"``, ``"not_for"``.
        executor: Optional :class:`AbstractToolExecutor` that runs the
            tool off-process (Kubernetes pod, Qworker, etc.). When
            ``None`` (the default), the tool executes in-process —
            identical to the legacy behaviour.
        webhook_callback_url: When set, the executor returns a
            ``ToolResult(status="pending")`` immediately and the
            final result arrives at this URL out-of-band. Only
            honoured by executors that support async delivery.
        remote_timeout_seconds: Max wall-clock seconds the executor
            waits for the remote runtime to return. Ignored when
            ``executor`` is ``None``.
        **kwargs: Additional configuration
    """
    # routing_meta — per-instance to avoid shared mutable default
    self.routing_meta: Dict = routing_meta if routing_meta is not None else {}

    # Remote execution wiring (None = legacy in-process behaviour)
    self.executor: Optional["AbstractToolExecutor"] = executor
    self.webhook_callback_url: Optional[str] = webhook_callback_url
    self.remote_timeout_seconds: int = int(remote_timeout_seconds)

    # Store initialization parameters for cloning. The live executor
    # instance is captured so clone() reuses the same transport;
    # ToolExecutionEnvelope serialization strips it before sending.
    self._init_kwargs = {
        'name': name,
        'description': description,
        'output_dir': output_dir,
        'base_url': base_url,
        'static_dir': static_dir,
        'executor': executor,
        'webhook_callback_url': webhook_callback_url,
        'remote_timeout_seconds': remote_timeout_seconds,
        **kwargs
    }

    # Set name and description
    self.name = name or self.name or self.__class__.__name__
    self.description = description or self.__class__.__doc__ or f"Tool: {self.name}"

    # Initialize permission context (per-call, set in execute())
    self._current_pctx: Optional[Any] = None

    # Set up logging
    self.logger = logging.getLogger(
        f'{self.name}.Tool'
    )

    # JSON encoders/decoders
    self._json_encoder = json_encoder
    self._json_decoder = json_decoder
    self._json = JSONContent()

    # File and URL configuration
    self.base_url = base_url or BASE_STATIC_URL
    self.static_url = base_url or BASE_STATIC_URL
    parsed = urlparse(self.static_url)
    self._base_scheme_netloc = (parsed.scheme, parsed.netloc)

    # Set up directories
    self.static_dir = Path(static_dir or STATIC_DIR).resolve()

    self.output_dir = Path(output_dir).resolve() if output_dir else self._default_output_dir()

    # Ensure output directory exists if specified
    if self.output_dir and not self.output_dir.exists():
        self.output_dir.mkdir(parents=True, exist_ok=True)

    # FEAT-176: initialise per-instance event registry
    self._init_events()

AgentTool

AgentTool(agent: 'AbstractBot', tool_name: str = None, tool_description: str = None, use_conversation_method: bool = True, context_filter: Optional[Callable[[AgentContext], AgentContext]] = None, execution_memory: Optional[Any] = None)

Bases: AbstractTool

Wraps any BasicAgent/AbstractBot as a tool for use by other agents.

• Schema includes "parameters" key for Google GenAI compatibility
• Uses Pydantic args_schema for validation
• Accepts all args as **kwargs in _execute()

Source code in packages/ai-parrot/src/parrot/tools/agent.py

def __init__(
    self,
    agent: "AbstractBot",
    tool_name: str = None,
    tool_description: str = None,
    use_conversation_method: bool = True,
    context_filter: Optional[Callable[[AgentContext], AgentContext]] = None,
    execution_memory: Optional[Any] = None,
):

    self.agent = agent
    self.name = tool_name or f"{agent.name.lower().replace(' ', '_')}"

    # Build description
    if tool_description:
        self.description = tool_description or getattr(agent, 'description', f"Execute {agent.name} agent")
    else:
        # Auto-generate from agent properties
        desc_parts = []
        if hasattr(agent, 'role') and agent.role:
            desc_parts.append(f"Role: {agent.role}")
        if hasattr(agent, 'goal') and agent.goal:
            desc_parts.append(f"Goal: {agent.goal}")
        if hasattr(agent, 'capabilities') and agent.capabilities:
            desc_parts.append(f"Capabilities: {agent.capabilities}")

        if desc_parts:
            self.description = f"Agent: {agent.name}. " + ". ".join(desc_parts)
        else:
            self.description = f"Consult {agent.name} for specialized assistance"

    self.use_conversation_method = use_conversation_method
    self.context_filter = context_filter
    self.execution_memory = execution_memory

    # Track usage
    self.call_count = 0
    self.last_response = None

    super().__init__(
        name=self.name,
        description=self.description,
        args_schema=QuestionInput  # Uses the modified schema
    )

    # Build schema in the correct format for Google GenAI
    # CRITICAL: Must have "parameters" key at top level
    self._schema = {
        "name": self.name,
        "description": self.description,
        "parameters": {  # ← Google GenAI extracts this key
            "type": "object",
            "properties": {
                "question": {
                    "type": "string",
                    "description": f"The question or task to ask {agent.name}"
                },
                "include_previous_results": {
                    "type": "boolean",
                    "description": (
                        "When true, automatically inject all previous agent "
                        "results as context into this call for cross-pollination"
                    )
                }
            },
            "required": ["question"],
            "additionalProperties": False
        }
    }

get_schema

get_schema() -> Dict[str, Any]

Return the tool schema in the format expected by Google GenAI.

RETURNS	DESCRIPTION
Dict[str, Any]	Schema with structure:
Dict[str, Any]	{ "name": "tool_name", "description": "...", "parameters": { # ← Google GenAI looks for this "type": "object", "properties": {...}, "required": [...] }
Dict[str, Any]	}

Source code in packages/ai-parrot/src/parrot/tools/agent.py

def get_schema(self) -> Dict[str, Any]:
    """
    Return the tool schema in the format expected by Google GenAI.

    Returns:
        Schema with structure:
        {
            "name": "tool_name",
            "description": "...",
            "parameters": {  # ← Google GenAI looks for this
                "type": "object",
                "properties": {...},
                "required": [...]
            }
        }
    """
    return self._schema

get_usage_stats

get_usage_stats() -> Dict[str, Any]

Get usage statistics for this agent tool.

Source code in packages/ai-parrot/src/parrot/tools/agent.py

def get_usage_stats(self) -> Dict[str, Any]:
    """Get usage statistics for this agent tool."""
    return {
        'name': self.name,
        'agent_name': self.agent.name,
        'call_count': self.call_count,
        'last_response_length': len(self.last_response) if self.last_response else 0
    }

SpawnSubAgentTool

SpawnSubAgentTool(bot_manager: Any, owner_id: str, *, allowed_tools: Optional[List[str]] = None, name: str = 'spawn_sub_agent', description: Optional[str] = None, routing_meta: Optional[Dict[str, Any]] = None)

Bases: AbstractTool

Spawn an ephemeral sub-agent to execute a single task.

Creates a short-lived sub-agent owned by the calling agent, executes one task with a restricted tool subset and a timeout, then discards the sub-agent — regardless of success, error, or timeout.

The tool never calls promote_user_bot; all sub-agents are ephemeral and discarded after their task completes.

PARAMETER	DESCRIPTION
bot_manager	The BotManager instance (injected via constructor — testable without an aiohttp app). TYPE: Any
owner_id	Canonical string ID of the parent agent that owns the sub-agent (e.g. "agent:orchestrator-001"). TYPE: str
allowed_tools	Allowlist of tool names the parent authorises for sub-agents. The sub-agent receives only the intersection of this list and the tools requested in the call. TYPE: Optional[List[str]] DEFAULT: None
name	Tool name (default: "spawn_sub_agent"). TYPE: str DEFAULT: 'spawn_sub_agent'
description	Tool description override. TYPE: Optional[str] DEFAULT: None
routing_meta	Routing hints for the CapabilityRegistry. The key "requires_grant" is reserved for future HITL grant enforcement (FEAT-grants); set but not enforced here. TYPE: Optional[Dict[str, Any]] DEFAULT: None

Initialize SpawnSubAgentTool.

PARAMETER	DESCRIPTION
bot_manager	BotManager instance — injected, not taken from app["bot_manager"], so the tool is testable standalone. TYPE: Any
owner_id	Canonical owner ID of the parent agent. TYPE: str
allowed_tools	Parent-defined allowlist of tool names. Sub-agents may only use tools in this list. Empty list means no tools. TYPE: Optional[List[str]] DEFAULT: None
name	Tool name (default: "spawn_sub_agent"). TYPE: str DEFAULT: 'spawn_sub_agent'
description	Override description for the LLM tool descriptor. TYPE: Optional[str] DEFAULT: None
routing_meta	Optional routing hints. The "requires_grant" key is set to False by default (no enforcement yet). TYPE: Optional[Dict[str, Any]] DEFAULT: None

Source code in packages/ai-parrot/src/parrot/tools/spawn.py

def __init__(
    self,
    bot_manager: Any,
    owner_id: str,
    *,
    allowed_tools: Optional[List[str]] = None,
    name: str = "spawn_sub_agent",
    description: Optional[str] = None,
    routing_meta: Optional[Dict[str, Any]] = None,
) -> None:
    """Initialize SpawnSubAgentTool.

    Args:
        bot_manager: ``BotManager`` instance — injected, not taken from
            ``app["bot_manager"]``, so the tool is testable standalone.
        owner_id: Canonical owner ID of the parent agent.
        allowed_tools: Parent-defined allowlist of tool names.  Sub-agents
            may only use tools in this list.  Empty list means no tools.
        name: Tool name (default: ``"spawn_sub_agent"``).
        description: Override description for the LLM tool descriptor.
        routing_meta: Optional routing hints.  The ``"requires_grant"``
            key is set to ``False`` by default (no enforcement yet).
    """
    # Build routing_meta with requires_grant and requires_confirmation placeholders.
    effective_routing = dict(routing_meta or {})
    effective_routing.setdefault("requires_grant", False)
    effective_routing.setdefault("requires_confirmation", False)  # FEAT-235

    super().__init__(
        name=name,
        description=description or self.description,
        routing_meta=effective_routing,
    )

    self._bot_manager = bot_manager
    self._owner_id = owner_id
    self._allowed_tools: List[str] = list(allowed_tools or [])

SpawnSubAgentInput

Bases: BaseModel

Input schema for SpawnSubAgentTool.

ATTRIBUTE	DESCRIPTION
task	The question / task for the ephemeral sub-agent. TYPE: str
tools	Allowed tool names for the sub-agent. Intersected with the parent's allowed_tools allowlist for defense in depth. TYPE: List[str]
model	LLM model override. Inherits parent default when not set. TYPE: Optional[str]
system_prompt	System prompt injected into the sub-agent. TYPE: Optional[str]
timeout	Max seconds the sub-agent is allowed to run before the call is cancelled. Defaults to 120 s. TYPE: int
ttl_seconds	Ephemeral registry TTL. Keep short (default 300 s / 5 min) for sub-agents — they should be discarded well before this. TYPE: int

tool_schema

tool_schema(schema: Type[BaseModel], description: Optional[str] = None)

Decorator to specify a custom argument schema for a toolkit method.

Usage

@tool_schema(MyCustomSchema) async def my_tool(self, arg1: str, arg2: int) -> str: '''My custom tool.''' return result

Source code in packages/ai-parrot/src/parrot/tools/decorators.py

def tool_schema(schema: Type[BaseModel], description: Optional[str] = None):
    """
    Decorator to specify a custom argument schema for a toolkit method.

    Usage:
        @tool_schema(MyCustomSchema)
        async def my_tool(self, arg1: str, arg2: int) -> str:
            '''My custom tool.'''
            return result
    """
    def decorator(func):
        # print(f"Decorating {func.__name__} with {schema}")
        func._args_schema = schema
        func._tool_description = description or func.__doc__ or f"Tool: {func.__name__}"
        return func
    return decorator

tool

tool(_func: Optional[Callable] = None, *, name: Optional[str] = None, description: Optional[str] = None, schema: Optional[Dict[str, Any]] = None, auto_register: bool = False, requires_confirmation: bool = False, confirm_template: Optional[str] = None, confirm_window_seconds: int = 0, allow_edit: bool = False)

Decorator to mark a function as a tool with automatic schema generation.

Automatically extracts: - Name from function name (or use custom name) - Description from docstring (or use custom description) - Input schema from type hints (or use custom schema)

PARAMETER	DESCRIPTION
name	Optional custom tool name (defaults to function name) TYPE: Optional[str] DEFAULT: None
description	Optional custom description (defaults to docstring) TYPE: Optional[str] DEFAULT: None
schema	Optional custom input schema (auto-generated from type hints if not provided) TYPE: Optional[Dict[str, Any]] DEFAULT: None
auto_register	If True, automatically register with active client/bot TYPE: bool DEFAULT: False
requires_confirmation	If True, the tool requires HITL confirmation before execution (via ConfirmationGuard in ToolManager — FEAT-235). TYPE: bool DEFAULT: False
confirm_template	Optional Python format string for the briefing shown to the human. Placeholders: {tool} (tool name), {params} (all params as k=v), plus any individual parameter name. Falls back to a raw tool with: k=v listing when None or on a template error. TYPE: Optional[str] DEFAULT: None
confirm_window_seconds	Seconds during which an identical call (same tool, same args_hash) is skipped without re-asking. 0 (default) means always re-ask — the safe per-call default. TYPE: int DEFAULT: 0
allow_edit	When True, the human is offered a FORM interaction to edit the parameter values before approving. Edited values are re-validated against the tool's args_schema. TYPE: bool DEFAULT: False

Usage

@tool def get_weather(location: str) -> str: '''Get weather for a location.''' return f"Weather in {location}"

@tool() def get_weather(location: str) -> str: '''Get weather for a location.''' return f"Weather in {location}"

@tool(name="custom_name", description="Custom description") def my_function(param: int) -> str: return str(param)

@tool(requires_confirmation=True, confirm_template="Check in {employee_id}?", confirm_window_seconds=60, allow_edit=True) def workday_checkin(employee_id: int, time: str) -> str: '''Register a check-in.''' return "ok"

Source code in packages/ai-parrot/src/parrot/tools/decorators.py

def tool(
    _func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    schema: Optional[Dict[str, Any]] = None,
    auto_register: bool = False,
    requires_confirmation: bool = False,
    confirm_template: Optional[str] = None,
    confirm_window_seconds: int = 0,
    allow_edit: bool = False,
):
    """
    Decorator to mark a function as a tool with automatic schema generation.

    Automatically extracts:
    - Name from function name (or use custom name)
    - Description from docstring (or use custom description)
    - Input schema from type hints (or use custom schema)

    Args:
        name: Optional custom tool name (defaults to function name)
        description: Optional custom description (defaults to docstring)
        schema: Optional custom input schema (auto-generated from type hints if not provided)
        auto_register: If True, automatically register with active client/bot
        requires_confirmation: If True, the tool requires HITL confirmation before
            execution (via ConfirmationGuard in ToolManager — FEAT-235).
        confirm_template: Optional Python format string for the briefing shown to the
            human.  Placeholders: ``{tool}`` (tool name), ``{params}`` (all params as
            ``k=v``), plus any individual parameter name.  Falls back to a raw
            ``tool with: k=v`` listing when None or on a template error.
        confirm_window_seconds: Seconds during which an identical call (same tool,
            same args_hash) is skipped without re-asking.  ``0`` (default) means
            always re-ask — the safe per-call default.
        allow_edit: When True, the human is offered a FORM interaction to edit the
            parameter values before approving.  Edited values are re-validated against
            the tool's ``args_schema``.

    Usage:
        @tool
        def get_weather(location: str) -> str:
            '''Get weather for a location.'''
            return f"Weather in {location}"

        @tool()
        def get_weather(location: str) -> str:
            '''Get weather for a location.'''
            return f"Weather in {location}"

        @tool(name="custom_name", description="Custom description")
        def my_function(param: int) -> str:
            return str(param)

        @tool(requires_confirmation=True, confirm_template="Check in {employee_id}?",
              confirm_window_seconds=60, allow_edit=True)
        def workday_checkin(employee_id: int, time: str) -> str:
            '''Register a check-in.'''
            return "ok"
    """
    def decorator(func: Callable) -> Callable:
        # Extract metadata
        tool_name = name or func.__name__
        tool_description = description or _extract_description(func)

        # Generate schema from type hints if not provided
        if schema is None:
            tool_schema = _generate_schema_from_function(func)
        else:
            tool_schema = schema

        # Build routing_meta for HITL confirmation (FEAT-235)
        confirmation_routing: Dict[str, Any] = {
            "requires_confirmation": requires_confirmation,
            "confirm_window_seconds": confirm_window_seconds,
            "allow_edit": allow_edit,
        }
        if confirm_template is not None:
            confirmation_routing["confirm_template"] = confirm_template

        # Store metadata on the function
        func._tool_metadata = {
            'name': tool_name,
            'description': tool_description,
            'schema': tool_schema,
            'function': func,
            'auto_register': auto_register,
            'routing_meta': confirmation_routing,
        }

        # Mark as a tool
        func._is_tool = True

        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        # Preserve metadata on wrapper
        wrapper._tool_metadata = func._tool_metadata
        wrapper._is_tool = True

        return wrapper

    # Support both @tool and @tool() syntax
    if _func is not None:
        return decorator(_func)
    return decorator

get_supported_toolkits

get_supported_toolkits() -> Dict[str, Type[AbstractToolkit]]

Get the dictionary of supported toolkits.

Source code in packages/ai-parrot/src/parrot/tools/registry.py

def get_supported_toolkits() -> Dict[str, Type["AbstractToolkit"]]:
    """Get the dictionary of supported toolkits."""
    return ToolkitRegistry.get_registry()