Clients

Client for Interactions with LLMs (Language Models) This module provides a client interface for interacting with various LLMs. It includes functionality for sending requests, receiving responses, and handling errors.

AbstractClient

AbstractClient(preset: Optional[str] = None, tools: Optional[List[Union[str, AbstractTool]]] = None, use_tools: bool = False, debug: bool = True, tool_manager: Optional[ToolManager] = None, **kwargs)

Bases: EventEmitterMixin, ABC

Abstract base Class for LLM models.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def __init__(
    self,
    preset: Optional[str] = None,
    tools: Optional[List[Union[str, AbstractTool]]] = None,
    use_tools: bool = False,
    debug: bool = True,
    tool_manager: Optional[ToolManager] = None,
    **kwargs,
):
    self.__name__ = self.__class__.__name__
    self.model: str = kwargs.get("model", None)
    # Per-loop client cache: keyed by id(asyncio.get_running_loop()).
    # Each entry holds the SDK client instance and a weakref to the loop.
    self._clients_by_loop: dict[int, _LoopClientEntry] = {}
    # Per-loop locks: asyncio.Lock() is loop-bound, so one per loop.
    self._locks_by_loop: dict[int, asyncio.Lock] = {}
    self.session: Optional[aiohttp.ClientSession] = None
    self.use_session: bool = kwargs.get("use_session", self.use_session)
    if preset:
        preset_config = LLM_PRESETS.get(preset, LLM_PRESETS["default"])
        # define temp, top_k, top_p, max_tokens from selected preset:
        self.temperature = preset_config.get("temperature", 0.4)
        self.top_k = preset_config.get("top_k", 30)
        self.top_p = preset_config.get("top_p", 0.2)
        self.max_tokens = preset_config.get("max_tokens")
    else:
        # define default values from preset default:
        self.temperature = kwargs.get("temperature", 0)
        self.top_k = kwargs.get("top_k", 30)
        self.top_p = kwargs.get("top_p", 0.2)
        # ``None`` means "not configured" — the per-client
        # _default_max_tokens / _invoke_max_tokens defaults take over. Do
        # NOT reintroduce a literal here: a framework-wide default assigned
        # eagerly is indistinguishable from a deliberate caller choice, and
        # would shadow every per-client default downstream.
        self.max_tokens = kwargs.get("max_tokens")
    # Whether max_tokens was EXPLICITLY configured by the caller (directly
    # or via a preset) rather than left unset above. invoke() only honours
    # an explicit value, so that a client's _invoke_max_tokens default is
    # never shadowed by ask()'s generic one.
    self._max_tokens_configured: bool = "max_tokens" in kwargs or preset is not None
    # Per-instance override for invoke()'s output-token budget specifically.
    # ``None`` means "fall back to an explicit self.max_tokens, then the
    # class default" (see _resolve_max_tokens).
    self.invoke_max_tokens: Optional[int] = kwargs.get("invoke_max_tokens", None)
    self.base_headers.update(kwargs.get("headers", {}))
    self.api_key = kwargs.get("api_key", None)
    self.version = kwargs.get("version", self.version)
    self._config = config
    self.logger: logging.Logger = logging.getLogger(self.__name__)
    self._json: Any = JSONContent()
    self.client_type: str = kwargs.get("client_type", self.client_type)
    self._debug: bool = debug
    self._program: str = kwargs.get("program", "parrot")  # Default program slug
    # Use provided tool_manager or create a new one
    # This allows Agent to pass its tool_manager as a reference
    if tool_manager is not None:
        self._tool_manager = tool_manager
    else:
        self._tool_manager = ToolManager(logger=self.logger, debug=self._debug)
    self.tools: Dict[str, Union[ToolDefinition, AbstractTool]] = {}
    self.enable_tools: bool = use_tools
    # FEAT-438 G5: only create an instance attribute when the caller
    # explicitly passed fallback_model= (including explicit None) —
    # an unconditional assignment here would shadow a subclass's
    # class-level _fallback_model default on every instance (the
    # class attribute declared above, :~272, is the fallback when no
    # instance attribute is created).
    if "fallback_model" in kwargs:
        self._fallback_model = kwargs.pop("fallback_model")
    # Initialize tools if provided
    if use_tools and tools:
        self._tool_manager.default_tools(tools)
        self.enable_tools = True
    # FEAT-176: initialise per-instance lifecycle event registry.
    # forward_to_global=False so client events stay isolated unless the
    # owning bot wires them up via a parent trace.
    self._init_events(forward_to_global=False)

tool_manager property writable

tool_manager: ToolManager

Get the tool manager.

client property writable

client: Optional[Any]

Return the SDK client bound to the current event loop, or None.

The cache key is id(asyncio.get_running_loop()). Returns None when called outside a running loop (e.g. in __init__ or sync code).

RETURNS	DESCRIPTION
Optional[Any]	The loop-local SDK client instance, or None if no client has
Optional[Any]	been created yet for the current loop.

default_model property

default_model: str

Return the default model for the client.

get_client abstractmethod async

get_client() -> Any

Return the client instance.

Source code in packages/ai-parrot/src/parrot/clients/base.py

@abstractmethod
async def get_client(self) -> Any:
    """Return the client instance."""
    raise NotImplementedError

complete async

complete(prompt: str, *, model: Optional[str] = None, system_prompt: Optional[str] = None, max_tokens: Optional[int] = None, temperature: Optional[float] = None) -> str

Send a prompt, return the model's textual reply as a plain string.

Thin convenience wrapper around ask() for single-shot, tool-less text generation. Use cases:

• Interop with components that expect async complete(prompt) -> str (e.g. parrot_tools.scraping.PlanGenerator).
• Quick prompts where you don't need the full AIMessage.

Handles two things ask() does NOT:

1. Auto-enters the async context manager if the client isn't already initialized, so callers don't need async with client: before calling complete(). If the client was already entered (e.g. inside a surrounding async with), we reuse it and don't tear it down.
2. Extracts plain text from the response — ask() returns an AIMessage pydantic model (provider-specific clients) or a MessageResponse TypedDict with a content-block list. Both shapes collapse to a single string here.

PARAMETER	DESCRIPTION
prompt	User prompt. TYPE: str
model	Override the instance default model. TYPE: Optional[str] DEFAULT: None
system_prompt	Optional system prompt. TYPE: Optional[str] DEFAULT: None
max_tokens	Override default max tokens. TYPE: Optional[int] DEFAULT: None
temperature	Override default sampling temperature. TYPE: Optional[float] DEFAULT: None

RETURNS	DESCRIPTION
str	The model's textual response.

RAISES	DESCRIPTION
RuntimeError	If the response has no extractable text.

Source code in packages/ai-parrot/src/parrot/clients/base.py

async def complete(
    self,
    prompt: str,
    *,
    model: Optional[str] = None,
    system_prompt: Optional[str] = None,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
) -> str:
    """Send a prompt, return the model's textual reply as a plain string.

    Thin convenience wrapper around ``ask()`` for single-shot,
    tool-less text generation. Use cases:

    - Interop with components that expect ``async complete(prompt) -> str``
      (e.g. ``parrot_tools.scraping.PlanGenerator``).
    - Quick prompts where you don't need the full ``AIMessage``.

    Handles two things ``ask()`` does NOT:

    1. **Auto-enters the async context manager** if the client isn't
       already initialized, so callers don't need ``async with client:``
       before calling ``complete()``. If the client was already entered
       (e.g. inside a surrounding ``async with``), we reuse it and
       don't tear it down.
    2. **Extracts plain text** from the response — ``ask()`` returns
       an ``AIMessage`` pydantic model (provider-specific clients) or
       a ``MessageResponse`` TypedDict with a content-block list.
       Both shapes collapse to a single string here.

    Args:
        prompt: User prompt.
        model: Override the instance default model.
        system_prompt: Optional system prompt.
        max_tokens: Override default max tokens.
        temperature: Override default sampling temperature.

    Returns:
        The model's textual response.

    Raises:
        RuntimeError: If the response has no extractable text.
    """
    need_enter = self.client is None
    if need_enter:
        await self.__aenter__()
    try:
        resolved_model = (
            model or self.model or getattr(self, "default_model", None) or getattr(self, "_default_model", None)
        )
        kwargs: Dict[str, Any] = {"prompt": prompt, "model": resolved_model}
        if system_prompt is not None:
            kwargs["system_prompt"] = system_prompt
        if max_tokens is not None:
            kwargs["max_tokens"] = max_tokens
        if temperature is not None:
            kwargs["temperature"] = temperature
        response = await self.ask(**kwargs)
    finally:
        if need_enter:
            await self.__aexit__(None, None, None)

    text = self._extract_text(response)
    if not text:
        raise RuntimeError(f"LLM returned no extractable text " f"(response type: {type(response).__name__})")
    return text

close async

close() -> None

Close all per-loop SDK clients.

Delegates to close_all(). Both close and close_all exist so callers can be explicit about intent (spec §8 Q1).

Source code in packages/ai-parrot/src/parrot/clients/base.py

async def close(self) -> None:
    """Close all per-loop SDK clients.

    Delegates to ``close_all()``.  Both ``close`` and ``close_all`` exist
    so callers can be explicit about intent (spec §8 Q1).
    """
    await self.close_all()

close_all async

close_all() -> None

Tear down every per-loop SDK client entry.

Safely handles dead / foreign loops: entries whose loop has been garbage-collected or belongs to a different running context are dropped without awaiting their close() coroutine.

After this call, _clients_by_loop and _locks_by_loop are empty.

Source code in packages/ai-parrot/src/parrot/clients/base.py

async def close_all(self) -> None:
    """Tear down every per-loop SDK client entry.

    Safely handles dead / foreign loops: entries whose loop has been
    garbage-collected or belongs to a different running context are dropped
    without awaiting their ``close()`` coroutine.

    After this call, ``_clients_by_loop`` and ``_locks_by_loop`` are empty.
    """
    current = self._get_current_loop()
    current_id = id(current) if current is not None else None
    for loop_id, entry in list(self._clients_by_loop.items()):
        await self._safe_close_entry(entry, is_current_loop=(loop_id == current_id))
    self._clients_by_loop.clear()
    self._locks_by_loop.clear()

set_program

set_program(program_slug: str) -> None

Set the program slug for the client.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def set_program(self, program_slug: str) -> None:
    """Set the program slug for the client."""
    self._program = program_slug

set_tools

set_tools(tools: List[Union[str, AbstractTool]]) -> None

Set complete list of tools, replacing existing.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def set_tools(self, tools: List[Union[str, AbstractTool]]) -> None:
    """Set complete list of tools, replacing existing."""
    self.tool_manager.clear_tools()
    self.tools.clear()
    self.register_tools(tools)

get_tool

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

Get a tool by name from ToolManager or legacy tools.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def get_tool(self, name: str) -> Optional[AbstractTool]:
    """Get a tool by name from ToolManager or legacy tools."""
    # Try ToolManager first
    if tool := self.tool_manager.get_tool(name):
        return tool

    # Fall back to legacy tools
    legacy_tool = self.tools.get(name)
    return legacy_tool if isinstance(legacy_tool, AbstractTool) else None

register_tool

register_tool(tool: Union[ToolDefinition, AbstractTool] = None, name: str = None, description: str = None, input_schema: Dict[str, Any] = None, function: Callable = None) -> None

Register a Python function as a tool for LLM to call.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def register_tool(
    self,
    tool: Union[ToolDefinition, AbstractTool] = None,
    name: str = None,
    description: str = None,
    input_schema: Dict[str, Any] = None,
    function: Callable = None,
) -> None:
    """Register a Python function as a tool for LLM to call."""
    self.tool_manager.register_tool(
        tool=tool, name=name, description=description, input_schema=input_schema, function=function
    )

register_tools

register_tools(tools: List[Union[ToolDefinition, AbstractTool]]) -> None

Register multiple tools at once.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def register_tools(self, tools: List[Union[ToolDefinition, AbstractTool]]) -> None:
    """Register multiple tools at once."""
    self.tool_manager.register_tools(tools)
    self.enable_tools = True

register_python_tool

register_python_tool(report_dir: Optional[Path] = None) -> PythonREPLTool

Register Python REPL tool with a ClaudeAPIClient.

PARAMETER	DESCRIPTION
client	The ClaudeAPIClient instance
report_dir	Directory for saving reports TYPE: Optional[Path] DEFAULT: None

RETURNS	DESCRIPTION
PythonREPLTool	The PythonREPLTool instance

Source code in packages/ai-parrot/src/parrot/clients/base.py

def register_python_tool(
    self,
    report_dir: Optional[Path] = None,
) -> PythonREPLTool:
    """Register Python REPL tool with a ClaudeAPIClient.

    Args:
        client: The ClaudeAPIClient instance
        report_dir: Directory for saving reports

    Returns:
        The PythonREPLTool instance
    """
    if "python_repl" in self.tools:
        return self.tools["python_repl"]

    tool = PythonREPLTool(
        report_dir=report_dir,
        debug=self._debug,
    )
    self.tool_manager.add_tool(tool)
    return tool

list_tools

list_tools() -> List[str]

Get a list of all registered tool names.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def list_tools(self) -> List[str]:
    """Get a list of all registered tool names."""
    tool_names = self.tool_manager.list_tools()
    legacy_names = list(self.tools.keys())
    return tool_names + [name for name in legacy_names if name not in tool_names]

remove_tool

remove_tool(name: str) -> bool

Remove a tool by name.

PARAMETER	DESCRIPTION
name	Tool name to remove TYPE: str

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

Source code in packages/ai-parrot/src/parrot/clients/base.py

def remove_tool(self, name: str) -> bool:
    """
    Remove a tool by name.

    Args:
        name: Tool name to remove

    Returns:
        True if tool was removed, False if not found
    """
    self.tool_manager.remove_tool(name)

clear_tools

clear_tools() -> None

Clear all registered tools.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def clear_tools(self) -> None:
    """Clear all registered tools."""
    self.tool_manager.clear_tools()
    self.tools.clear()
    self.logger.info("Cleared all tools")

ask abstractmethod async

ask(prompt: str, model: str, max_tokens: Optional[int] = None, temperature: float = 0.7, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[str] = None, history: Optional[Sequence[HistoryMessage]] = None, structured_output: Union[type, StructuredOutputConfig, None] = None, tools: Optional[List[Dict[str, Any]]] = None, use_tools: Optional[bool] = None, deep_research: bool = False, background: bool = False, lazy_loading: bool = False) -> MessageResponse

Send a prompt to the model and return the response.

PARAMETER	DESCRIPTION
prompt	The input prompt for the model TYPE: str
model	The model to use TYPE: str
max_tokens	Maximum number of tokens in the response. None (the default) resolves via :meth:_resolve_max_tokens — the per-instance max_tokens, then the client's :attr:_default_max_tokens. TYPE: Optional[int] DEFAULT: None
temperature	Sampling temperature for response generation TYPE: float DEFAULT: 0.7
files	Optional files to include in the request TYPE: Optional[List[Union[str, Path]]] DEFAULT: None
system_prompt	Optional system prompt to guide the model TYPE: Optional[str] DEFAULT: None
history	Already-rendered conversation history from the owning bot (FEAT-524). The client only formats it for its provider — it never loads or persists history itself. TYPE: Optional[Sequence[HistoryMessage]] DEFAULT: None
structured_output	Optional structured output configuration TYPE: Union[type, StructuredOutputConfig, None] DEFAULT: None
tools	Optional tools to register for this call TYPE: Optional[List[Dict[str, Any]]] DEFAULT: None
use_tools	Whether to use tools TYPE: Optional[bool] DEFAULT: None
deep_research	If True, use deep research mode (provider-specific) TYPE: bool DEFAULT: False
background	If True, execute research in background (async mode) TYPE: bool DEFAULT: False
lazy_loading	If True, enabled dynamic tool searching TYPE: bool DEFAULT: False

Source code in packages/ai-parrot/src/parrot/clients/base.py

@abstractmethod
async def ask(
    self,
    prompt: str,
    model: str,
    max_tokens: Optional[int] = None,
    temperature: float = 0.7,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[str] = None,
    history: Optional[Sequence[HistoryMessage]] = None,
    structured_output: Union[type, StructuredOutputConfig, None] = None,
    tools: Optional[List[Dict[str, Any]]] = None,
    use_tools: Optional[bool] = None,
    deep_research: bool = False,
    background: bool = False,
    lazy_loading: bool = False,
) -> MessageResponse:
    """Send a prompt to the model and return the response.

    Args:
        prompt: The input prompt for the model
        model: The model to use
        max_tokens: Maximum number of tokens in the response. ``None`` (the
            default) resolves via :meth:`_resolve_max_tokens` — the
            per-instance ``max_tokens``, then the client's
            :attr:`_default_max_tokens`.
        temperature: Sampling temperature for response generation
        files: Optional files to include in the request
        system_prompt: Optional system prompt to guide the model
        history: Already-rendered conversation history from the owning bot
            (FEAT-524). The client only formats it for its provider — it
            never loads or persists history itself.
        structured_output: Optional structured output configuration
        tools: Optional tools to register for this call
        use_tools: Whether to use tools
        deep_research: If True, use deep research mode (provider-specific)
        background: If True, execute research in background (async mode)
        lazy_loading: If True, enabled dynamic tool searching
    """
    raise NotImplementedError("Subclasses must implement this method.")

ask_stream abstractmethod async

ask_stream(prompt: str, model: str = None, max_tokens: Optional[int] = None, temperature: float = 0.7, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[str] = None, history: Optional[Sequence[HistoryMessage]] = None, tools: Optional[List[Dict[str, Any]]] = None, deep_research: bool = False, agent_config: Optional[Dict[str, Any]] = None, lazy_loading: bool = False) -> AsyncIterator[Union[str, AIMessage]]

Stream the model's response.

Yields successive string chunks of the model response followed by a single final :class:~parrot.models.responses.AIMessage carrying full response metadata (token usage, stop reason, model, provider, turn_id, etc.).

Implementors MUST yield at least one str chunk before the final AIMessage. Consumers can detect the end-of-stream sentinel via isinstance(chunk, AIMessage).

Source code in packages/ai-parrot/src/parrot/clients/base.py

@abstractmethod
async def ask_stream(
    self,
    prompt: str,
    model: str = None,
    max_tokens: Optional[int] = None,
    temperature: float = 0.7,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[str] = None,
    history: Optional[Sequence[HistoryMessage]] = None,
    tools: Optional[List[Dict[str, Any]]] = None,
    deep_research: bool = False,
    agent_config: Optional[Dict[str, Any]] = None,
    lazy_loading: bool = False,
) -> AsyncIterator[Union[str, AIMessage]]:
    """Stream the model's response.

    Yields successive string chunks of the model response followed by a
    single final :class:`~parrot.models.responses.AIMessage` carrying full
    response metadata (token usage, stop reason, model, provider, turn_id,
    etc.).

    Implementors MUST yield at least one ``str`` chunk before the final
    ``AIMessage``.  Consumers can detect the end-of-stream sentinel via
    ``isinstance(chunk, AIMessage)``.
    """
    raise NotImplementedError("Subclasses must implement this method.")

resume abstractmethod async

resume(session_id: str, user_input: str, state: Dict[str, Any]) -> MessageResponse

Resume a suspended model execution.

PARAMETER	DESCRIPTION
session_id	The session ID TYPE: str
user_input	The user's input to inject as tool result TYPE: str
state	The suspended state containing messages and tool_call_id TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
MessageResponse	The response from the LLM TYPE: MessageResponse

Source code in packages/ai-parrot/src/parrot/clients/base.py

@abstractmethod
async def resume(self, session_id: str, user_input: str, state: Dict[str, Any]) -> MessageResponse:
    """Resume a suspended model execution.

    Args:
        session_id: The session ID
        user_input: The user's input to inject as tool result
        state: The suspended state containing messages and tool_call_id

    Returns:
        MessageResponse: The response from the LLM
    """
    raise NotImplementedError("Subclasses must implement this method.")

batch_ask async

batch_ask(requests: List[Any]) -> List[Any]

Process multiple requests in batch.

Source code in packages/ai-parrot/src/parrot/clients/base.py

async def batch_ask(self, requests: List[Any]) -> List[Any]:
    """Process multiple requests in batch."""
    raise NotImplementedError("Subclasses must implement batch processing.")

invoke abstractmethod async

invoke(prompt: str, *, output_type: Optional[type] = None, structured_output: Optional[StructuredOutputConfig] = None, model: Optional[str] = None, system_prompt: Optional[str] = None, max_tokens: Optional[int] = None, temperature: float = 0.0, use_tools: bool = False, tools: Optional[list] = None) -> InvokeResult

Lightweight stateless invocation — no retry, no history, no prompt builder.

Each concrete client implements this method using provider-native structured output. Use this instead of ask() when you need fast, stateless structured extraction without conversation history overhead.

PARAMETER	DESCRIPTION
prompt	The user prompt to send. TYPE: str
output_type	A Pydantic model or dataclass class to parse the response into. Mutually exclusive with structured_output (structured_output wins). TYPE: Optional[type] DEFAULT: None
structured_output	Full :class:StructuredOutputConfig, including optional custom_parser. Takes precedence over output_type. TYPE: Optional[StructuredOutputConfig] DEFAULT: None
model	Override the model for this call. Falls back to an explicitly selected self.model, then _lightweight_model, then _default_model (see :meth:_resolve_invoke_model). TYPE: Optional[str] DEFAULT: None
system_prompt	Override the system prompt. Falls back to BASIC_SYSTEM_PROMPT rendered with instance attributes. TYPE: Optional[str] DEFAULT: None
max_tokens	Maximum completion tokens. None (the default) resolves via :meth:_resolve_invoke_max_tokens — per-instance invoke_max_tokens, then max_tokens, then the client's :attr:_invoke_max_tokens class default. Pass an explicit value only to cap a single call. TYPE: Optional[int] DEFAULT: None
temperature	Sampling temperature (default 0.0 for deterministic output). TYPE: float DEFAULT: 0.0
use_tools	If True, inject registered tools into the request. TYPE: bool DEFAULT: False
tools	Additional tool definitions to pass directly to the provider. TYPE: Optional[list] DEFAULT: None

RETURNS	DESCRIPTION
InvokeResult	class:InvokeResult with the parsed output, model, usage,
InvokeResult	and optional raw_response.

RAISES	DESCRIPTION
	class:InvokeError: If the provider call fails for any reason.

Source code in packages/ai-parrot/src/parrot/clients/base.py

@abstractmethod
async def invoke(
    self,
    prompt: str,
    *,
    output_type: Optional[type] = None,
    structured_output: Optional[StructuredOutputConfig] = None,
    model: Optional[str] = None,
    system_prompt: Optional[str] = None,
    max_tokens: Optional[int] = None,
    temperature: float = 0.0,
    use_tools: bool = False,
    tools: Optional[list] = None,
) -> InvokeResult:
    """Lightweight stateless invocation — no retry, no history, no prompt builder.

    Each concrete client implements this method using provider-native structured
    output. Use this instead of ``ask()`` when you need fast, stateless structured
    extraction without conversation history overhead.

    Args:
        prompt: The user prompt to send.
        output_type: A Pydantic model or dataclass class to parse the response into.
            Mutually exclusive with ``structured_output`` (``structured_output`` wins).
        structured_output: Full :class:`StructuredOutputConfig`, including optional
            ``custom_parser``. Takes precedence over ``output_type``.
        model: Override the model for this call. Falls back to an
            explicitly selected ``self.model``, then ``_lightweight_model``,
            then ``_default_model`` (see :meth:`_resolve_invoke_model`).
        system_prompt: Override the system prompt. Falls back to
            ``BASIC_SYSTEM_PROMPT`` rendered with instance attributes.
        max_tokens: Maximum completion tokens. ``None`` (the default) resolves
            via :meth:`_resolve_invoke_max_tokens` — per-instance
            ``invoke_max_tokens``, then ``max_tokens``, then the client's
            :attr:`_invoke_max_tokens` class default. Pass an explicit value
            only to cap a single call.
        temperature: Sampling temperature (default 0.0 for deterministic output).
        use_tools: If ``True``, inject registered tools into the request.
        tools: Additional tool definitions to pass directly to the provider.

    Returns:
        :class:`InvokeResult` with the parsed ``output``, ``model``, ``usage``,
        and optional ``raw_response``.

    Raises:
        :class:`InvokeError`: If the provider call fails for any reason.
    """
    raise NotImplementedError("Subclasses must implement invoke().")

StreamingRetryConfig

StreamingRetryConfig(max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, backoff_factor: float = 2.0, jitter: bool = True, auto_retry_on_max_tokens: bool = True, token_increase_factor: float = 1.5, retry_on_rate_limit: bool = True, retry_on_server_error: bool = True)

Configuration for streaming retry behavior.

Source code in packages/ai-parrot/src/parrot/clients/base.py

def __init__(
    self,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff_factor: float = 2.0,
    jitter: bool = True,
    auto_retry_on_max_tokens: bool = True,
    token_increase_factor: float = 1.5,
    retry_on_rate_limit: bool = True,
    retry_on_server_error: bool = True,
):
    self.max_retries = max_retries
    self.base_delay = base_delay
    self.max_delay = max_delay
    self.backoff_factor = backoff_factor
    self.jitter = jitter
    self.auto_retry_on_max_tokens = auto_retry_on_max_tokens
    self.token_increase_factor = token_increase_factor
    self.retry_on_rate_limit = retry_on_rate_limit
    self.retry_on_server_error = retry_on_server_error

OpenAIBaseClient

OpenAIBaseClient(api_key: str | None = None, base_url: str | None = None, **kwargs)

Bases: AbstractClient

OpenAI-compatible wire protocol; carries NO OpenAI-provider defaults.

Subclasses that speak the OpenAI chat-completions wire protocol under a provider-specific label (Bedrock Mantle, OpenRouter, Moonshot, Nvidia, LocalLLM/vLLM, and — in Phase 2 — Groq/Zai via their native SDKs) should inherit from this class instead of :class:~parrot.clients.gpt.OpenAIClient directly, so they never inherit OpenAI-the-provider defaults (OpenAI-only model ids, Responses-API routing, Sora, etc.).

Initialize the OpenAI-compatible wire client.

PARAMETER	DESCRIPTION
api_key	Bearer token for the target endpoint. Providers supply their own environment-variable default in their own __init__ — this base does not read any env var. TYPE: str | None DEFAULT: None
base_url	Base URL of the OpenAI-compatible endpoint. Providers supply their own default in their own __init__. TYPE: str | None DEFAULT: None
**kwargs	Forwarded to :class:~parrot.clients.base.AbstractClient. May include model (normalized via :meth:_normalize_model before being forwarded) and timeout (SDK request timeout, defaulting to this class's :attr:_default_timeout, which subclasses raise for slower endpoints). DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

def __init__(
    self,
    api_key: str | None = None,
    base_url: str | None = None,
    **kwargs,
):
    """Initialize the OpenAI-compatible wire client.

    Args:
        api_key: Bearer token for the target endpoint. Providers supply
            their own environment-variable default in their own
            ``__init__`` — this base does not read any env var.
        base_url: Base URL of the OpenAI-compatible endpoint. Providers
            supply their own default in their own ``__init__``.
        **kwargs: Forwarded to :class:`~parrot.clients.base.AbstractClient`.
            May include ``model`` (normalized via :meth:`_normalize_model`
            before being forwarded) and ``timeout`` (SDK request timeout,
            defaulting to this class's :attr:`_default_timeout`, which
            subclasses raise for slower endpoints).
    """
    self.api_key = api_key
    self.base_url = base_url
    self.base_headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {self.api_key}",
    }
    self._timeout = kwargs.pop("timeout", self._default_timeout)
    if "model" in kwargs:
        kwargs["model"] = self._normalize_model(kwargs["model"])
    super().__init__(**kwargs)

get_client async

get_client() -> Any

Build the default OpenAI-SDK-shaped async client.

Lazily imports openai.AsyncOpenAI so the SDK is only required when an OpenAI-compatible client is actually instantiated. Subclasses that wrap a native SDK (Groq, Zai) override this hook.

RETURNS	DESCRIPTION
Any	An AsyncOpenAI instance configured with this client's
Any	api_key/base_url/timeout.

RAISES	DESCRIPTION
ImportError	If the openai package is not installed.

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def get_client(self) -> Any:
    """Build the default OpenAI-SDK-shaped async client.

    Lazily imports ``openai.AsyncOpenAI`` so the SDK is only required
    when an OpenAI-compatible client is actually instantiated. Subclasses
    that wrap a native SDK (Groq, Zai) override this hook.

    Returns:
        An ``AsyncOpenAI`` instance configured with this client's
        ``api_key``/``base_url``/timeout.

    Raises:
        ImportError: If the ``openai`` package is not installed.
    """
    try:
        from openai import AsyncOpenAI
    except ImportError as exc:
        raise ImportError(
            "OpenAIBaseClient requires the 'openai' SDK. " "Install with: pip install ai-parrot[openai]"
        ) from exc
    return AsyncOpenAI(
        api_key=self.api_key,
        base_url=self.base_url,
        timeout=self._timeout,
    )

ask async

ask(prompt: str, model: Any | None = None, max_tokens: int | None = None, temperature: float | None = None, files: list[str | Path] | None = None, system_prompt: str | None = None, history: Sequence[HistoryMessage] | None = None, structured_output: type | StructuredOutputConfig | None = None, tools: list[dict[str, Any]] | None = None, use_tools: bool | None = None, lazy_loading: bool = False) -> AIMessage

Ask the OpenAI-compatible endpoint a question with optional conversation memory.

Generic chat-completions implementation shared by every OpenAIBaseClient subclass that speaks the plain OpenAI wire protocol. Subclasses needing Responses-API routing, deep-research dispatch, or other OpenAI-provider-only behavior override this method (currently only :class:~parrot.clients.gpt.OpenAIClient, which reuses :meth:_chat_completion/:meth:_run_tool_call_loop from this base while keeping its own richer ask()).

PARAMETER	DESCRIPTION
prompt	The prompt to send to the model. TYPE: str
model	The model to use, or None to use the configured one. TYPE: Any | None DEFAULT: None
max_tokens	Maximum tokens for the response. TYPE: int | None DEFAULT: None
temperature	Sampling temperature. TYPE: float | None DEFAULT: None
files	Files to upload before the call. TYPE: list[str | Path] | None DEFAULT: None
system_prompt	System prompt to prepend. TYPE: str | None DEFAULT: None
structured_output	Structured output definition (Pydantic model, dataclass, or explicit StructuredOutputConfig). TYPE: type | StructuredOutputConfig | None DEFAULT: None
user_id	User ID for conversation memory.
session_id	Session ID for conversation memory.
tools	Tools to register for this call. TYPE: list[dict[str, Any]] | None DEFAULT: None
use_tools	Whether to use tools; defaults to self.enable_tools. TYPE: bool | None DEFAULT: None
lazy_loading	If True, enable dynamic tool searching. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
AIMessage	The response from the model.

RAISES	DESCRIPTION
NotImplementedError	If the resolved model requires Responses-API routing (:meth:_is_responses_model) and this class does not override ask() to handle it.

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def ask(
    self,
    prompt: str,
    model: Any | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    files: list[str | Path] | None = None,
    system_prompt: str | None = None,
    history: Sequence[HistoryMessage] | None = None,
    structured_output: type | StructuredOutputConfig | None = None,
    tools: list[dict[str, Any]] | None = None,
    use_tools: bool | None = None,
    lazy_loading: bool = False,
) -> AIMessage:
    """Ask the OpenAI-compatible endpoint a question with optional conversation memory.

    Generic chat-completions implementation shared by every
    ``OpenAIBaseClient`` subclass that speaks the plain OpenAI wire
    protocol. Subclasses needing Responses-API routing, deep-research
    dispatch, or other OpenAI-provider-only behavior override this
    method (currently only :class:`~parrot.clients.gpt.OpenAIClient`,
    which reuses :meth:`_chat_completion`/:meth:`_run_tool_call_loop`
    from this base while keeping its own richer ``ask()``).

    Args:
        prompt: The prompt to send to the model.
        model: The model to use, or ``None`` to use the configured one.
        max_tokens: Maximum tokens for the response.
        temperature: Sampling temperature.
        files: Files to upload before the call.
        system_prompt: System prompt to prepend.
        structured_output: Structured output definition (Pydantic model,
            dataclass, or explicit ``StructuredOutputConfig``).
        user_id: User ID for conversation memory.
        session_id: Session ID for conversation memory.
        tools: Tools to register for this call.
        use_tools: Whether to use tools; defaults to ``self.enable_tools``.
        lazy_loading: If ``True``, enable dynamic tool searching.

    Returns:
        The response from the model.

    Raises:
        NotImplementedError: If the resolved model requires Responses-API
            routing (:meth:`_is_responses_model`) and this class does not
            override ``ask()`` to handle it.
    """
    turn_id = str(uuid.uuid4())
    original_prompt = prompt
    _use_tools = use_tools if use_tools is not None else self.enable_tools

    model_str = self._resolve_model(model)

    if self._is_responses_model(model_str):
        raise NotImplementedError(
            f"{type(self).__name__} does not implement Responses-API routing "
            "for this model; override ask() to handle it."
        )

    messages = self._build_messages(prompt, files, history)
    _lc_tc = self._emit_before_call(
        client_name=self.client_name,
        model=model_str,
        temperature=temperature if temperature is not None else self.temperature,
        system_prompt=system_prompt,
        has_tools=bool(_use_tools),
        parent_trace=None,
    )
    _lc_t0 = time.perf_counter()

    if files:
        for file in files:
            if isinstance(file, str):
                file = Path(file)
            if isinstance(file, Path):
                await self._upload_file(file)

    if lazy_loading and system_prompt:
        system_prompt += (
            "\n\nYou have access to a library of tools. Use the 'search_tools' function to find relevant tools."
        )
    elif lazy_loading and not system_prompt:
        system_prompt = (
            "You have access to a library of tools. Use the 'search_tools' function to find relevant tools."
        )

    if system_prompt:
        if isinstance(system_prompt, list):
            system_prompt = "\n\n".join(s.text for s in system_prompt)
        messages.insert(0, {"role": "system", "content": system_prompt})

    # FEAT-524: do NOT append the current turn here. _build_messages()
    # already placed it last, after the rendered history; appending again
    # sent the prompt twice and produced two consecutive `user` messages,
    # which strict-alternation providers reject.

    output_config = self._get_structured_config(structured_output)

    if tools and isinstance(tools, list):
        for tool in tools:
            self.register_tool(tool)

    active_tool_names = set()
    prepared_tools = None

    if _use_tools:
        if lazy_loading:
            prepared_tools = self._prepare_lazy_tools()
            if prepared_tools:
                active_tool_names.add("search_tools")
        else:
            prepared_tools = self._prepare_tools()

    args: dict[str, Any] = {}
    if prepared_tools:
        args["tools"] = prepared_tools
        args["tool_choice"] = "auto"

    _resolved_max_tokens = self._resolve_max_tokens(max_tokens)
    if _resolved_max_tokens is not None:
        args["max_tokens"] = _resolved_max_tokens
    if temperature:
        args["temperature"] = temperature

    if output_config:
        args["response_format"] = self._build_response_format_from(output_config)

    _used_fallback = False
    _original_model = model_str

    _round_t0 = time.perf_counter()
    try:
        response = await self._chat_completion(model=model_str, messages=messages, use_tools=_use_tools, **args)
    except Exception as e:
        if self._should_use_fallback(model_str, e):
            self.logger.warning(
                "Model %s capacity error: %s. Retrying once with fallback: %s",
                model_str,
                e,
                self._fallback_model,
            )
            model_str = self._fallback_model
            _used_fallback = True
            response = await self._chat_completion(model=model_str, messages=messages, use_tools=_use_tools, **args)
        else:
            raise
    _round_duration_ms = (time.perf_counter() - _round_t0) * 1000
    result = response.choices[0].message

    def _on_round(round_number, usage, raw_usage, tool_names, duration_ms):
        self._emit_round_event(
            _lc_tc,
            client_name=self.client_name,
            model=model_str,
            round_number=round_number,
            usage=usage,
            raw_usage=raw_usage,
            tool_calls=tool_names,
            duration_ms=duration_ms,
        )

    result, response, all_tool_calls, accumulated_usage, round_number = await self._run_tool_call_loop(
        result=result,
        response=response,
        messages=messages,
        model_str=model_str,
        use_tools=_use_tools,
        args=args,
        session_id=current_session_id.get(),
        lazy_loading=lazy_loading,
        active_tool_names=active_tool_names,
        track_usage=True,
        initial_duration_ms=_round_duration_ms,
        on_round=_on_round,
    )

    messages.append({"role": "assistant", "content": result.content})

    response_text = result.content if isinstance(result.content, str) else self._json.dumps(result.content)
    final_output = None
    if output_config:
        try:
            # Known-truncated output must not reach a custom parser either.
            self._raise_if_truncated(self._extract_finish_reason(response), model=model_str)
            if output_config.custom_parser:
                final_output = output_config.custom_parser(response_text)
            else:
                final_output = await self._parse_structured_output(
                    response_text,
                    output_config,
                    finish_reason=self._extract_finish_reason(response),
                    model=model_str,
                )
        except InvokeError:
            raise
        except Exception:  # noqa: BLE001 pylint: disable=broad-except
            final_output = response_text

    # FEAT-524: no memory write — AbstractBot.save_conversation_turn is the single writer.

    structured_payload = None
    if final_output is not None and not (isinstance(final_output, str) and final_output == response_text):
        structured_payload = final_output

    ai_message = AIMessageFactory.from_openai(
        response=response,
        input_text=original_prompt,
        model=model_str,
        user_id=current_user_id.get(),
        session_id=current_session_id.get(),
        turn_id=turn_id,
        structured_output=structured_payload,
    )

    if accumulated_usage is not None:
        if round_number > 1:
            accumulated_usage.extra_usage["rounds"] = round_number
        ai_message.usage = accumulated_usage

    ai_message.tool_calls = all_tool_calls
    if _used_fallback:
        ai_message.metadata["used_fallback_model"] = True
        ai_message.metadata["original_model"] = _original_model
        ai_message.metadata["fallback_model"] = self._fallback_model

    _lc_usage = getattr(ai_message, "usage", None)
    await self._emit_after_call(
        _lc_tc,
        client_name=self.client_name,
        model=model_str,
        duration_ms=(time.perf_counter() - _lc_t0) * 1000,
        input_tokens=getattr(_lc_usage, "prompt_tokens", None) if _lc_usage else None,
        output_tokens=getattr(_lc_usage, "completion_tokens", None) if _lc_usage else None,
        finish_reason=getattr(ai_message, "stop_reason", None),
    )
    return ai_message

resume async

resume(session_id: str, user_input: str, state: dict[str, Any]) -> AIMessage

Resume a suspended model execution.

PARAMETER	DESCRIPTION
session_id	The session ID. TYPE: str
user_input	The user's input to inject as a tool result. TYPE: str
state	The suspended state containing messages and tool_call_id. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
AIMessage	The response from the model.

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def resume(self, session_id: str, user_input: str, state: dict[str, Any]) -> AIMessage:
    """Resume a suspended model execution.

    Args:
        session_id: The session ID.
        user_input: The user's input to inject as a tool result.
        state: The suspended state containing messages and tool_call_id.

    Returns:
        The response from the model.
    """
    await self._ensure_client()

    messages = state["messages"]
    tool_call_id = state["tool_call_id"]
    model_str = state.get("agent_name", self.model or self.default_model)

    messages.append({"role": "tool", "tool_call_id": tool_call_id, "name": "handoff_tool", "content": user_input})

    turn_id = str(uuid.uuid4())

    response = await self._chat_completion(model=model_str, messages=messages, use_tools=True)
    result = response.choices[0].message

    result, response, all_tool_calls, _accumulated_usage, _round_number = await self._run_tool_call_loop(
        result=result,
        response=response,
        messages=messages,
        model_str=model_str,
        use_tools=True,
        args={},
        session_id=session_id,
        record_malformed_tool_calls=False,
        default_tool_name="unknown",
    )

    ai_message = AIMessageFactory.from_openai(
        response=response,
        input_text="[Resumed Conversation]",
        model=model_str,
        user_id="unknown",
        session_id=session_id,
        turn_id=turn_id,
    )
    ai_message.tool_calls = all_tool_calls
    return ai_message

batch_ask async

batch_ask(requests: list[dict[str, Any]]) -> list[AIMessage]

Process multiple ask() requests sequentially.

No native batch API exists for the OpenAI wire protocol; requests are processed one at a time via :meth:ask.

PARAMETER	DESCRIPTION
requests	A list of kwargs dicts, each forwarded to :meth:ask. TYPE: list[dict[str, Any]]

RETURNS	DESCRIPTION
list[AIMessage]	The list of :class:AIMessage responses, in request order.

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def batch_ask(self, requests: list[dict[str, Any]]) -> list[AIMessage]:
    """Process multiple ``ask()`` requests sequentially.

    No native batch API exists for the OpenAI wire protocol; requests
    are processed one at a time via :meth:`ask`.

    Args:
        requests: A list of kwargs dicts, each forwarded to :meth:`ask`.

    Returns:
        The list of :class:`AIMessage` responses, in request order.
    """
    results = []
    for request in requests:
        result = await self.ask(**request)
        results.append(result)
    return results

ask_stream async

ask_stream(prompt: str, model: Any | None = None, max_tokens: int | None = None, temperature: float | None = None, files: list[str | Path] | None = None, system_prompt: str | None = None, history: Sequence[HistoryMessage] | None = None, tools: list[dict[str, Any]] | None = None, use_tools: bool = True, structured_output: type | StructuredOutputConfig | None = None, lazy_loading: bool = False, **kwargs) -> AsyncIterator[str | AIMessage]

Stream a response with tool-use support and conversation memory.

Generic chat-completions streaming implementation shared by every OpenAIBaseClient subclass. Subclasses needing Responses-API routing or other OpenAI-provider-only streaming behavior override this method (currently only :class:~parrot.clients.gpt.OpenAIClient). Routes through :meth:_chat_completion with stream=True (FEAT-438 G3 — the single completion funnel) instead of calling the SDK directly, so a subclass's funnel override applies to streaming too.

When the model requests tool calls during streaming, the tool-call chunks are accumulated, the tools executed between rounds, and a new streaming request is issued with the tool results — mirroring the _run_tool_call_loop() used by :meth:ask.

Yields successive string chunks followed by a final :class:~parrot.models.responses.AIMessage.

PARAMETER	DESCRIPTION
use_tools	Whether to include tool definitions and run the streaming tool-call loop. Defaults to True. TYPE: bool DEFAULT: True

RAISES	DESCRIPTION
NotImplementedError	If the resolved model requires Responses-API routing and this class does not override ask_stream().

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def ask_stream(
    self,
    prompt: str,
    model: Any | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    files: list[str | Path] | None = None,
    system_prompt: str | None = None,
    history: Sequence[HistoryMessage] | None = None,
    tools: list[dict[str, Any]] | None = None,
    use_tools: bool = True,
    structured_output: type | StructuredOutputConfig | None = None,
    lazy_loading: bool = False,
    **kwargs,
) -> AsyncIterator[str | AIMessage]:
    """Stream a response with tool-use support and conversation memory.

    Generic chat-completions streaming implementation shared by every
    ``OpenAIBaseClient`` subclass. Subclasses needing Responses-API
    routing or other OpenAI-provider-only streaming behavior override
    this method (currently only :class:`~parrot.clients.gpt.OpenAIClient`).
    Routes through :meth:`_chat_completion` with ``stream=True``
    (FEAT-438 G3 — the single completion funnel) instead of calling the
    SDK directly, so a subclass's funnel override applies to streaming
    too.

    When the model requests tool calls during streaming, the tool-call
    chunks are accumulated, the tools executed between rounds, and a new
    streaming request is issued with the tool results — mirroring the
    ``_run_tool_call_loop()`` used by :meth:`ask`.

    Yields successive string chunks followed by a final
    :class:`~parrot.models.responses.AIMessage`.

    Args:
        use_tools: Whether to include tool definitions and run the
            streaming tool-call loop.  Defaults to ``True``.

    Raises:
        NotImplementedError: If the resolved model requires Responses-API
            routing and this class does not override ``ask_stream()``.
    """
    turn_id = str(uuid.uuid4())
    model_str = self._resolve_model(model)

    if self._is_responses_model(model_str):
        raise NotImplementedError(
            f"{type(self).__name__} does not implement Responses-API "
            "streaming for this model; override ask_stream() to handle it."
        )

    messages = self._build_messages(prompt, files, history)
    _lc_tc = self._emit_before_call(
        client_name=self.client_name,
        model=model_str,
        temperature=temperature if temperature is not None else self.temperature,
        system_prompt=system_prompt,
        has_tools=bool(self.tools) and use_tools,
        parent_trace=None,
    )
    _lc_t0 = time.perf_counter()

    if files:
        for file in files:
            if isinstance(file, str):
                file = Path(file)
            if isinstance(file, Path):
                await self._upload_file(file)

    if lazy_loading and system_prompt:
        system_prompt += (
            "\n\nYou have access to a library of tools. Use the 'search_tools' function to find relevant tools."
        )
    elif lazy_loading and not system_prompt:
        system_prompt = (
            "You have access to a library of tools. Use the 'search_tools' function to find relevant tools."
        )

    if system_prompt:
        if isinstance(system_prompt, list):
            system_prompt = "\n\n".join(s.text for s in system_prompt)
        messages.insert(0, {"role": "system", "content": system_prompt})

    if tools and isinstance(tools, list):
        for tool in tools:
            self.register_tool(tool)

    tools_payload = None
    if use_tools and self.tools:
        if lazy_loading:
            tools_payload = self._prepare_lazy_tools()
        else:
            tools_payload = self._prepare_tools()

    args: dict[str, Any] = {"stream_options": {"include_usage": True}}
    if tools_payload:
        args["tools"] = tools_payload
        args["tool_choice"] = "auto"

    max_tokens_value = self._resolve_max_tokens(max_tokens)
    if max_tokens_value is not None:
        args["max_tokens"] = max_tokens_value

    temperature_value = temperature if temperature is not None else self.temperature
    if temperature_value is not None:
        args["temperature"] = temperature_value

    output_config = self._get_structured_config(structured_output)
    if output_config:
        response_format = self._build_response_format_from(output_config)
        if response_format:
            args["response_format"] = response_format

    # ── Streaming tool-call loop ──────────────────────────────────
    # Mirrors _run_tool_call_loop() used by ask(). Each round streams
    # text and accumulates tool-call chunks; when the model finishes
    # with tool calls, they are executed between rounds and the
    # conversation continues.
    all_tool_calls: list[ToolCall] = []
    assistant_content = ""
    usage_data = None
    _max_tool_rounds = 25  # safety cap

    for _round in range(_max_tool_rounds):
        # `.parse()` cannot reliably stream every SDK's tool-calling/plain
        # responses; only prefer it when structured output was requested —
        # mirrors the pre-FEAT-438 dispatch this funnel now formalizes.
        response_stream = await self._chat_completion(
            model=model_str, messages=messages, use_tools=not bool(output_config), stream=True, **args
        )

        # Accumulate tool-call chunks by index. OpenAI streams them
        # incrementally: first chunk carries id+name, subsequent ones
        # carry argument fragments.
        _tc_accum: dict[int, dict[str, str]] = {}
        _round_content = ""
        _finish_reason: str | None = None

        async for chunk in response_stream:
            if chunk.choices:
                choice = chunk.choices[0]
                delta = choice.delta
                if delta and delta.content:
                    text_chunk = delta.content
                    _round_content += text_chunk
                    assistant_content += text_chunk
                    yield text_chunk
                # Accumulate streamed tool-call fragments.
                if delta and getattr(delta, "tool_calls", None):
                    for tc_delta in delta.tool_calls:
                        idx = tc_delta.index
                        if idx not in _tc_accum:
                            _tc_accum[idx] = {
                                "id": getattr(tc_delta, "id", None) or "",
                                "name": "",
                                "arguments": "",
                            }
                        if tc_delta.function:
                            if tc_delta.function.name:
                                _tc_accum[idx]["name"] = tc_delta.function.name
                            if tc_delta.function.arguments:
                                _tc_accum[idx]["arguments"] += tc_delta.function.arguments
                _finish_reason = getattr(choice, "finish_reason", None) or _finish_reason
            if hasattr(chunk, "usage") and chunk.usage is not None:
                usage_data = chunk.usage

        # ── Tool-use round: execute and loop ──────────────────────
        if _tc_accum and _finish_reason in ("tool_calls", "stop"):
            # Build assistant message with tool_calls for conversation.
            tc_serialized = []
            for idx in sorted(_tc_accum):
                tc_info = _tc_accum[idx]
                tc_serialized.append(
                    {
                        "id": tc_info["id"],
                        "type": "function",
                        "function": {
                            "name": tc_info["name"],
                            "arguments": tc_info["arguments"],
                        },
                    }
                )
            messages.append(
                {
                    "role": "assistant",
                    "content": _round_content or None,
                    "tool_calls": tc_serialized,
                }
            )

            for idx in sorted(_tc_accum):
                tc_info = _tc_accum[idx]
                try:
                    tool_args = json.loads(tc_info["arguments"]) if tc_info["arguments"] else {}
                except (json.JSONDecodeError, TypeError):
                    tool_args = {}
                tc = ToolCall(
                    id=tc_info["id"],
                    name=tc_info["name"],
                    arguments=tool_args,
                )
                try:
                    start_time = time.time()
                    tool_result = await self._execute_tool(tc_info["name"], tool_args)
                    tc.result = tool_result
                    tc.execution_time = time.time() - start_time
                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": tc_info["id"],
                            "name": tc_info["name"],
                            "content": str(tool_result),
                        }
                    )
                except Exception as e:
                    from parrot.core.exceptions import HumanInteractionInterrupt

                    if isinstance(e, HumanInteractionInterrupt):
                        e.session_id = current_session_id.get()
                        e.messages = messages
                        e.tool_call_id = tc_info["id"]
                        e.agent_name = model_str
                        raise
                    tc.error = str(e)
                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": tc_info["id"],
                            "name": tc_info["name"],
                            "content": str(e),
                        }
                    )
                all_tool_calls.append(tc)
            # Continue to next streaming round.
        else:
            # No tool calls — append assistant content and break.
            if _round_content:
                messages.append({"role": "assistant", "content": _round_content})
            break

    if usage_data is not None:
        usage = CompletionUsage.from_openai(usage_data)
    else:
        usage = CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)

    ai_message = AIMessage(
        input=prompt,
        output=assistant_content,
        response=assistant_content,
        model=model_str,
        provider=self.client_type,
        usage=usage,
        user_id=current_user_id.get(),
        session_id=current_session_id.get(),
        turn_id=turn_id,
        tool_calls=all_tool_calls,
    )
    await self._emit_after_call(
        _lc_tc,
        client_name=self.client_name,
        model=model_str,
        duration_ms=(time.perf_counter() - _lc_t0) * 1000,
        input_tokens=getattr(usage, "prompt_tokens", None),
        output_tokens=getattr(usage, "completion_tokens", None),
        finish_reason=_finish_reason,
    )
    yield ai_message

invoke async

invoke(prompt: str, *, output_type: type | None = None, structured_output: StructuredOutputConfig | None = None, model: str | None = None, system_prompt: str | None = None, max_tokens: int | None = None, temperature: float = 0.0, use_tools: bool = False, tools: list | None = None) -> InvokeResult

Lightweight stateless invocation routed through the completion funnel.

Generic implementation shared by every OpenAIBaseClient subclass. A single call is made through :meth:_chat_completion (FEAT-438 G3 — previously this bypassed the funnel entirely and called the SDK directly) — no conversation history, no prompt builder; use_tools=True is passed to the funnel so it always uses .create() (matching the pre-FEAT-438 behavior of never using .parse() here).

PARAMETER	DESCRIPTION
prompt	User prompt. TYPE: str
output_type	Pydantic model or dataclass to parse the response into. TYPE: type | None DEFAULT: None
structured_output	Full StructuredOutputConfig; takes precedence over output_type. TYPE: StructuredOutputConfig | None DEFAULT: None
model	Model override. Falls back to an explicitly selected self.model, then _lightweight_model, then _default_model. TYPE: str | None DEFAULT: None
system_prompt	System prompt override. TYPE: str | None DEFAULT: None
max_tokens	Maximum completion tokens. TYPE: int | None DEFAULT: None
temperature	Sampling temperature. TYPE: float DEFAULT: 0.0
use_tools	Whether to inject registered tools. TYPE: bool DEFAULT: False
tools	Additional tool definitions (unused — kept for interface parity with :class:~parrot.clients.gpt.OpenAIClient). TYPE: list | None DEFAULT: None

RETURNS	DESCRIPTION
InvokeResult	class:InvokeResult with parsed output.

RAISES	DESCRIPTION
InvokeError	On provider errors, or if the client is not initialized.

Source code in packages/ai-parrot/src/parrot/clients/openai_base.py

async def invoke(
    self,
    prompt: str,
    *,
    output_type: type | None = None,
    structured_output: StructuredOutputConfig | None = None,
    model: str | None = None,
    system_prompt: str | None = None,
    max_tokens: int | None = None,
    temperature: float = 0.0,
    use_tools: bool = False,
    tools: list | None = None,
) -> InvokeResult:
    """Lightweight stateless invocation routed through the completion funnel.

    Generic implementation shared by every ``OpenAIBaseClient``
    subclass. A single call is made through :meth:`_chat_completion`
    (FEAT-438 G3 — previously this bypassed the funnel entirely and
    called the SDK directly) — no conversation history, no prompt
    builder; ``use_tools=True`` is passed to the funnel so it always
    uses ``.create()`` (matching the pre-FEAT-438 behavior of never
    using ``.parse()`` here).

    Args:
        prompt: User prompt.
        output_type: Pydantic model or dataclass to parse the response into.
        structured_output: Full ``StructuredOutputConfig``; takes
            precedence over ``output_type``.
        model: Model override. Falls back to an explicitly selected
            ``self.model``, then ``_lightweight_model``, then
            ``_default_model``.
        system_prompt: System prompt override.
        max_tokens: Maximum completion tokens.
        temperature: Sampling temperature.
        use_tools: Whether to inject registered tools.
        tools: Additional tool definitions (unused — kept for interface
            parity with :class:`~parrot.clients.gpt.OpenAIClient`).

    Returns:
        :class:`InvokeResult` with parsed output.

    Raises:
        InvokeError: On provider errors, or if the client is not
            initialized.
    """
    try:
        resolved_prompt = self._resolve_invoke_system_prompt(system_prompt)
        config = self._build_invoke_structured_config(output_type, structured_output)
        resolved_model = self._resolve_invoke_model(model)
        max_tokens = self._resolve_max_tokens(max_tokens, resolved_model, for_invoke=True)

        messages = [
            {"role": "system", "content": resolved_prompt},
            {"role": "user", "content": prompt},
        ]

        kwargs: dict[str, Any] = {
            "max_tokens": max_tokens,
            "temperature": temperature,
        }

        if config:
            response_format = self._build_response_format_from(config)
            if response_format:
                kwargs["response_format"] = response_format

        if use_tools:
            tool_defs = self._prepare_tools()
            if tool_defs:
                kwargs["tools"] = tool_defs

        if not self.client:
            raise RuntimeError(f"{type(self).__name__} not initialised. Use async context manager.")

        response = await self._chat_completion(model=resolved_model, messages=messages, use_tools=True, **kwargs)
        raw_text = response.choices[0].message.content or ""

        output: Any = raw_text
        if config:
            # Known-truncated output must not reach a custom parser either.
            self._raise_if_truncated(self._extract_finish_reason(response), model=resolved_model)
            if config.custom_parser:
                output = config.custom_parser(raw_text)
            else:
                output = await self._parse_structured_output(
                    raw_text,
                    config,
                    finish_reason=self._extract_finish_reason(response),
                    model=resolved_model,
                )

        usage = CompletionUsage.from_openai(response.usage)
        return self._build_invoke_result(output, output_type, resolved_model, usage, response)
    except InvokeError:
        raise
    except Exception as exc:  # noqa: BLE001 — funnel errors are wrapped into InvokeError
        raise self._handle_invoke_error(exc)

ZaiClient

ZaiClient(api_key: Optional[str] = None, base_url: str = 'https://api.z.ai/api/paas/v4/', timeout: Optional[float] = None, max_retries: Optional[int] = None, **kwargs: Any)

Bases: OpenAIBaseClient

Client for Z.ai chat completions using the official zai-sdk package.

FEAT-438 (TASK-2304): rebased onto OpenAIBaseClient. The inherited tool_format = ToolFormat.OPENAI is CORRECT and left undeclared here — Z.ai's API takes the same {"type":"function","function":{...}} wrapper — but it is a non-issue either way: ask()/ask_stream()/ resume()/invoke() build tool payloads via this module's own _prepare_zai_tools() (kept, never calls the inherited _prepare_tools()), which never emits "strict" — so the base's OPENAI-gated strict-tools branch (base.py:1435) never applies to any real Z.ai request regardless of the declared tool_format.

Unlike Groq's AsyncGroq, the official zai SDK is synchronous — every wire call wraps client.chat.completions.create in asyncio.to_thread() (see _chat_completion below, adapted from the pre-rebase _create_completion/_stream_completion seams into the shared funnel signature from TASK-2298).

Source code in packages/ai-parrot/src/parrot/clients/zai.py

def __init__(
    self,
    api_key: Optional[str] = None,
    base_url: str = "https://api.z.ai/api/paas/v4/",
    timeout: Optional[float] = None,
    max_retries: Optional[int] = None,
    **kwargs: Any,
) -> None:
    resolved_key = api_key or config.get("ZAI_API_KEY")
    if not resolved_key:
        raise ValueError("ZAI_API_KEY is required. Pass api_key= or set the ZAI_API_KEY environment variable.")
    resolved_base_url = base_url or config.get("ZAI_BASE_URL") or "https://api.z.ai/api/paas/v4/"
    self.timeout = timeout
    self.max_retries = max_retries
    super().__init__(
        api_key=resolved_key,
        base_url=resolved_base_url,
        **kwargs,
    )
    # Re-set after super().__init__ because AbstractClient may overwrite
    # self.api_key during its own initialisation. This mirrors the
    # guard used by NvidiaClient/OpenRouterClient/MoonshotClient/
    # GroqClient.
    self.api_key = resolved_key

get_client async

get_client() -> Any

Create the official Z.ai SDK client for the current event loop.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def get_client(self) -> Any:
    """Create the official Z.ai SDK client for the current event loop."""
    from zai import ZaiClient as OfficialZaiClient

    kwargs: Dict[str, Any] = {
        "api_key": self.api_key,
        "base_url": self.base_url,
    }
    if self.timeout is not None:
        kwargs["timeout"] = self.timeout
    if self.max_retries is not None:
        kwargs["max_retries"] = self.max_retries
    return OfficialZaiClient(**kwargs)

ask async

ask(prompt: str, model: Union[str, ZaiModel, None] = None, max_tokens: Optional[int] = None, temperature: float = 0.7, top_p: float = 0.9, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[Union[str, list]] = None, history: Optional[Sequence[HistoryMessage]] = None, structured_output: Union[type, StructuredOutputConfig, None] = None, tools: Optional[List[Dict[str, Any]]] = None, use_tools: Optional[bool] = None, thinking: Optional[Union[bool, str, Dict[str, Any]]] = None, deep_thinking: bool = False, **_: Any) -> AIMessage

Send a non-streaming chat request to Z.ai.

PARAMETER	DESCRIPTION
prompt	The user input text. TYPE: str
model	Z.ai model identifier; defaults to :attr:_default_model. TYPE: Union[str, ZaiModel, None] DEFAULT: None
max_tokens	Maximum completion tokens. TYPE: Optional[int] DEFAULT: None
temperature	Sampling temperature. TYPE: float DEFAULT: 0.7
top_p	Top-p nucleus sampling parameter. TYPE: float DEFAULT: 0.9
files	Optional file paths to include in the request. TYPE: Optional[List[Union[str, Path]]] DEFAULT: None
system_prompt	Optional system prompt string or list of CacheableSegments. TYPE: Optional[Union[str, list]] DEFAULT: None
structured_output	Pydantic model or :class:StructuredOutputConfig for JSON-schema-constrained responses. TYPE: Union[type, StructuredOutputConfig, None] DEFAULT: None
user_id	Optional user identifier for conversation memory.
session_id	Optional session identifier for conversation memory.
tools	Additional tool definitions to register for this call. TYPE: Optional[List[Dict[str, Any]]] DEFAULT: None
use_tools	Override the instance-level enable_tools flag. TYPE: Optional[bool] DEFAULT: None
thinking	Enable chain-of-thought for thinking-capable models. TYPE: Optional[Union[bool, str, Dict[str, Any]]] DEFAULT: None
deep_thinking	Shorthand to enable thinking on capable models. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
AIMessage	class:AIMessage with the final response and usage metadata.

RAISES	DESCRIPTION
Exception	Propagates provider errors after emitting a ClientCallFailedEvent.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def ask(
    self,
    prompt: str,
    model: Union[str, ZaiModel, None] = None,
    max_tokens: Optional[int] = None,
    temperature: float = 0.7,
    top_p: float = 0.9,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[Union[str, list]] = None,
    history: Optional[Sequence[HistoryMessage]] = None,
    structured_output: Union[type, StructuredOutputConfig, None] = None,
    tools: Optional[List[Dict[str, Any]]] = None,
    use_tools: Optional[bool] = None,
    thinking: Optional[Union[bool, str, Dict[str, Any]]] = None,
    deep_thinking: bool = False,
    **_: Any,
) -> AIMessage:
    """Send a non-streaming chat request to Z.ai.

    Args:
        prompt: The user input text.
        model: Z.ai model identifier; defaults to :attr:`_default_model`.
        max_tokens: Maximum completion tokens.
        temperature: Sampling temperature.
        top_p: Top-p nucleus sampling parameter.
        files: Optional file paths to include in the request.
        system_prompt: Optional system prompt string or list of
            CacheableSegments.
        structured_output: Pydantic model or :class:`StructuredOutputConfig`
            for JSON-schema-constrained responses.
        user_id: Optional user identifier for conversation memory.
        session_id: Optional session identifier for conversation memory.
        tools: Additional tool definitions to register for this call.
        use_tools: Override the instance-level ``enable_tools`` flag.
        thinking: Enable chain-of-thought for thinking-capable models.
        deep_thinking: Shorthand to enable thinking on capable models.

    Returns:
        :class:`AIMessage` with the final response and usage metadata.

    Raises:
        Exception: Propagates provider errors after emitting a
            ``ClientCallFailedEvent``.
    """
    max_tokens = self._resolve_max_tokens(max_tokens)
    resolved_model = self._model_value(model)
    max_tokens = self._resolve_max_tokens(max_tokens, resolved_model, for_invoke=True)
    turn_id = str(uuid.uuid4())
    started = time.perf_counter()
    messages, resolved_system_prompt = self._build_zai_messages(
        prompt,
        files,
        history,
        system_prompt,
    )

    _use_tools = use_tools if use_tools is not None else self.enable_tools
    if tools:
        for tool in tools:
            self.register_tool(tool)

    output_config = self._get_structured_config(structured_output)
    request_args: Dict[str, Any] = {
        "model": resolved_model,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "top_p": top_p,
        "stream": False,
    }

    if thinking_payload := self._thinking_payload(resolved_model, thinking, deep_thinking):
        request_args["thinking"] = thinking_payload

    if _use_tools:
        request_args["tools"] = self._prepare_zai_tools()
        request_args["tool_choice"] = "auto"
    elif output_config:
        self._ensure_json_instruction(
            messages,
            "Please respond with a valid JSON object that matches the requested schema.",
        )
        request_args.update(
            self._prepare_structured_output_format(output_config.output_type)
            if output_config.format == OutputFormat.JSON
            else {}
        )

    # FEAT-176/228: emit before-call lifecycle event
    lc_tc = self._emit_before_call(
        client_name="zai",
        model=resolved_model,
        temperature=temperature,
        system_prompt=resolved_system_prompt,
        has_tools=bool(_use_tools),
    )
    try:
        response = await self._chat_completion(**request_args, use_tools=_use_tools)
        all_tool_calls: List[ToolCall] = []
        if _use_tools:
            response, all_tool_calls = await self._run_tool_loop(
                messages=messages,
                response=response,
                request_args=request_args,
            )

        content = getattr(response.choices[0].message, "content", None) or ""
        parsed_output = None
        if output_config:
            parsed_output = await self._parse_structured_output(
                content,
                output_config,
                finish_reason=self._extract_finish_reason(response),
                model=resolved_model,
            )

        response_time = time.perf_counter() - started
        ai_message = self._create_ai_message(
            response=response,
            input_text=prompt,
            model=resolved_model,
            user_id=current_user_id.get(),
            session_id=current_session_id.get(),
            turn_id=turn_id,
            structured_output=parsed_output,
            tool_calls=all_tool_calls,
            response_time=response_time,
        )
    except Exception as exc:
        await self._emit_failed_call(
            lc_tc,
            client_name="zai",
            model=resolved_model,
            duration_ms=(time.perf_counter() - started) * 1000,
            exc=exc,
        )
        raise

    await self._emit_after_call(
        lc_tc,
        client_name="zai",
        model=resolved_model,
        duration_ms=response_time * 1000,
        input_tokens=ai_message.usage.prompt_tokens,
        output_tokens=ai_message.usage.completion_tokens,
        finish_reason=ai_message.stop_reason,
    )
    # FEAT-524: no memory write — AbstractBot.save_conversation_turn is the single writer.
    return ai_message

ask_stream async

ask_stream(prompt: str, model: Union[str, ZaiModel, None] = None, max_tokens: Optional[int] = None, temperature: float = 0.7, top_p: float = 0.9, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[Union[str, list]] = None, history: Optional[Sequence[HistoryMessage]] = None, tools: Optional[List[Dict[str, Any]]] = None, use_tools: Optional[bool] = None, thinking: Optional[Union[bool, str, Dict[str, Any]]] = None, deep_thinking: bool = False, stream_reasoning: bool = False, **_: Any) -> AsyncIterator[Union[str, AIMessage]]

Stream a Z.ai response, yielding text chunks followed by an :class:AIMessage sentinel.

PARAMETER	DESCRIPTION
prompt	The user input text. TYPE: str
model	Z.ai model identifier; defaults to :attr:_default_model. TYPE: Union[str, ZaiModel, None] DEFAULT: None
max_tokens	Maximum completion tokens. TYPE: Optional[int] DEFAULT: None
temperature	Sampling temperature. TYPE: float DEFAULT: 0.7
top_p	Top-p nucleus sampling parameter. TYPE: float DEFAULT: 0.9
files	Optional file paths to include in the request. TYPE: Optional[List[Union[str, Path]]] DEFAULT: None
system_prompt	Optional system prompt string or list of CacheableSegments. TYPE: Optional[Union[str, list]] DEFAULT: None
user_id	Optional user identifier for conversation memory.
session_id	Optional session identifier for conversation memory.
tools	Additional tool definitions to register for this call. TYPE: Optional[List[Dict[str, Any]]] DEFAULT: None
use_tools	Override the instance-level enable_tools flag. TYPE: Optional[bool] DEFAULT: None
thinking	Enable chain-of-thought for thinking-capable models. TYPE: Optional[Union[bool, str, Dict[str, Any]]] DEFAULT: None
deep_thinking	Shorthand to enable thinking on capable models. TYPE: bool DEFAULT: False
stream_reasoning	When True, yield reasoning-content chunks as they arrive in addition to the final content. TYPE: bool DEFAULT: False

YIELDS	DESCRIPTION
AsyncIterator[Union[str, AIMessage]]	str chunks of the response as they arrive, followed by a
single	class:AIMessage sentinel carrying full metadata. TYPE:: AsyncIterator[Union[str, AIMessage]]

RAISES	DESCRIPTION
Exception	Propagates provider errors after emitting a ClientCallFailedEvent.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def ask_stream(
    self,
    prompt: str,
    model: Union[str, ZaiModel, None] = None,
    max_tokens: Optional[int] = None,
    temperature: float = 0.7,
    top_p: float = 0.9,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[Union[str, list]] = None,
    history: Optional[Sequence[HistoryMessage]] = None,
    tools: Optional[List[Dict[str, Any]]] = None,
    use_tools: Optional[bool] = None,
    thinking: Optional[Union[bool, str, Dict[str, Any]]] = None,
    deep_thinking: bool = False,
    stream_reasoning: bool = False,
    **_: Any,
) -> AsyncIterator[Union[str, AIMessage]]:
    """Stream a Z.ai response, yielding text chunks followed by an
    :class:`AIMessage` sentinel.

    Args:
        prompt: The user input text.
        model: Z.ai model identifier; defaults to :attr:`_default_model`.
        max_tokens: Maximum completion tokens.
        temperature: Sampling temperature.
        top_p: Top-p nucleus sampling parameter.
        files: Optional file paths to include in the request.
        system_prompt: Optional system prompt string or list of
            CacheableSegments.
        user_id: Optional user identifier for conversation memory.
        session_id: Optional session identifier for conversation memory.
        tools: Additional tool definitions to register for this call.
        use_tools: Override the instance-level ``enable_tools`` flag.
        thinking: Enable chain-of-thought for thinking-capable models.
        deep_thinking: Shorthand to enable thinking on capable models.
        stream_reasoning: When ``True``, yield reasoning-content chunks as
            they arrive in addition to the final content.

    Yields:
        ``str`` chunks of the response as they arrive, followed by a
        single :class:`AIMessage` sentinel carrying full metadata.

    Raises:
        Exception: Propagates provider errors after emitting a
            ``ClientCallFailedEvent``.
    """
    max_tokens = self._resolve_max_tokens(max_tokens)
    resolved_model = self._model_value(model)
    turn_id = str(uuid.uuid4())
    started = time.perf_counter()
    messages, resolved_system_prompt = self._build_zai_messages(
        prompt,
        files,
        history,
        system_prompt,
    )

    _use_tools = use_tools if use_tools is not None else self.enable_tools
    if tools:
        for tool in tools:
            self.register_tool(tool)

    request_args: Dict[str, Any] = {
        "model": resolved_model,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "top_p": top_p,
        "stream": True,
    }
    if thinking_payload := self._thinking_payload(resolved_model, thinking, deep_thinking):
        request_args["thinking"] = thinking_payload
    if _use_tools:
        request_args["tools"] = self._prepare_zai_tools()
        request_args["tool_choice"] = "auto"
        request_args["tool_stream"] = True

    # FEAT-176/228: emit before-call lifecycle event
    lc_tc = self._emit_before_call(
        client_name="zai",
        model=resolved_model,
        temperature=temperature,
        system_prompt=resolved_system_prompt,
        has_tools=bool(_use_tools),
    )

    content_parts: List[str] = []
    reasoning_parts: List[str] = []
    usage = CompletionUsage()
    finish_reason: Optional[str] = None
    last_raw_chunk: Dict[str, Any] = {}
    tool_call_accumulator: Dict[int, Dict[str, Any]] = {}

    try:
        async for chunk in await self._chat_completion(**request_args, use_tools=_use_tools):
            last_raw_chunk = self._response_to_dict(chunk)
            if getattr(chunk, "usage", None):
                usage = self._usage_from_response(chunk)
            if not getattr(chunk, "choices", None):
                continue
            choice = chunk.choices[0]
            finish_reason = getattr(choice, "finish_reason", None) or finish_reason
            delta = getattr(choice, "delta", None)
            if delta is None:
                continue
            reasoning = getattr(delta, "reasoning_content", None)
            if reasoning:
                reasoning_parts.append(reasoning)
                if stream_reasoning:
                    yield reasoning
            content = getattr(delta, "content", None)
            if content:
                content_parts.append(content)
                yield content
            self._accumulate_stream_tool_calls(
                tool_call_accumulator,
                getattr(delta, "tool_calls", None),
            )

        all_tool_calls: List[ToolCall] = []
        if tool_call_accumulator:
            assistant_tool_calls = [tool_call_accumulator[index] for index in sorted(tool_call_accumulator)]
            messages.append(
                {
                    "role": "assistant",
                    "content": "".join(content_parts),
                    "tool_calls": assistant_tool_calls,
                }
            )
            for provider_tool_call in assistant_tool_calls:
                function = provider_tool_call["function"]
                tool_name = function["name"]
                tool_args = self._parse_tool_arguments(function["arguments"])
                tool_call = ToolCall(
                    id=provider_tool_call.get("id") or str(uuid.uuid4()),
                    name=tool_name,
                    arguments=tool_args,
                )
                try:
                    tool_started = time.perf_counter()
                    tool_result = await self._execute_tool(tool_name, tool_args)
                    tool_call.execution_time = time.perf_counter() - tool_started
                    tool_call.result = tool_result
                    tool_content = json.dumps(tool_result, default=str)
                except Exception as exc:
                    tool_call.error = str(exc)
                    tool_content = f"Error: {exc}"
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "name": tool_name,
                        "content": tool_content,
                    }
                )
                all_tool_calls.append(tool_call)

            follow_up_args = dict(request_args)
            follow_up_args["messages"] = messages
            follow_up_args.pop("tool_stream", None)
            async for chunk in await self._chat_completion(**follow_up_args, use_tools=True):
                last_raw_chunk = self._response_to_dict(chunk)
                if getattr(chunk, "usage", None):
                    usage = self._usage_from_response(chunk)
                if not getattr(chunk, "choices", None):
                    continue
                choice = chunk.choices[0]
                finish_reason = getattr(choice, "finish_reason", None) or finish_reason
                delta = getattr(choice, "delta", None)
                if delta is None:
                    continue
                content = getattr(delta, "content", None)
                if content:
                    content_parts.append(content)
                    yield content

        content_text = "".join(content_parts)
        if not content_text:
            yield ""

        metadata: Dict[str, Any] = {}
        reasoning_text = "".join(reasoning_parts)
        if reasoning_text:
            metadata["reasoning_content"] = reasoning_text
        if usage.extra_usage.get("cached_tokens") is not None:
            metadata["cached_tokens"] = usage.extra_usage["cached_tokens"]

        response_time = time.perf_counter() - started
        ai_message = AIMessage(
            input=prompt,
            output=content_text,
            response=content_text,
            model=resolved_model,
            provider="zai",
            usage=usage,
            stop_reason=finish_reason,
            finish_reason=finish_reason,
            tool_calls=all_tool_calls,
            user_id=current_user_id.get(),
            session_id=current_session_id.get(),
            turn_id=turn_id,
            response_time=response_time,
            raw_response=last_raw_chunk,
            metadata=metadata,
        )
    except Exception as exc:
        await self._emit_failed_call(
            lc_tc,
            client_name="zai",
            model=resolved_model,
            duration_ms=(time.perf_counter() - started) * 1000,
            exc=exc,
        )
        raise

    await self._emit_after_call(
        lc_tc,
        client_name="zai",
        model=resolved_model,
        duration_ms=response_time * 1000,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        finish_reason=finish_reason,
    )
    # FEAT-524: no memory write — AbstractBot.save_conversation_turn is the single writer.
    yield ai_message

resume async

resume(session_id: str, user_input: str, state: Dict[str, Any]) -> AIMessage

Resume a suspended ZaiClient execution after a HandoffTool / HITL pause.

Injects user_input into the suspended message history (as a tool role message when state["tool_call_id"] is present, otherwise as a user message) and continues the tool-call loop until a final response is produced.

PARAMETER	DESCRIPTION
session_id	Session identifier propagated to any :class:~parrot.core.exceptions.HumanInteractionInterrupt raised inside the loop. TYPE: str
user_input	User reply to inject as the resumption value. TYPE: str
state	Suspended execution state. Expected keys: messages (list): OpenAI-style message dicts. tool_call_id (str, optional): ID of the paused tool call. When present user_input is injected as a tool result; otherwise it is injected as a user turn. model / agent_name (str, optional): Model override. user_id (str, optional): Propagated to the returned :class:AIMessage. TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
AIMessage	class:AIMessage with the final assistant response and all tool
AIMessage	calls executed during resumption.

RAISES	DESCRIPTION
	class:~parrot.core.exceptions.HumanInteractionInterrupt: Re-raised with updated session context when a tool triggers another human-interaction pause.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def resume(
    self,
    session_id: str,
    user_input: str,
    state: Dict[str, Any],
) -> AIMessage:
    """Resume a suspended ZaiClient execution after a HandoffTool / HITL pause.

    Injects *user_input* into the suspended message history (as a ``tool``
    role message when ``state["tool_call_id"]`` is present, otherwise as a
    ``user`` message) and continues the tool-call loop until a final
    response is produced.

    Args:
        session_id: Session identifier propagated to any
            :class:`~parrot.core.exceptions.HumanInteractionInterrupt`
            raised inside the loop.
        user_input: User reply to inject as the resumption value.
        state: Suspended execution state.  Expected keys:

            - ``messages`` (``list``): OpenAI-style message dicts.
            - ``tool_call_id`` (``str``, optional): ID of the paused tool
              call.  When present *user_input* is injected as a ``tool``
              result; otherwise it is injected as a ``user`` turn.
            - ``model`` / ``agent_name`` (``str``, optional): Model
              override.
            - ``user_id`` (``str``, optional): Propagated to the returned
              :class:`AIMessage`.

    Returns:
        :class:`AIMessage` with the final assistant response and all tool
        calls executed during resumption.

    Raises:
        :class:`~parrot.core.exceptions.HumanInteractionInterrupt`:
            Re-raised with updated session context when a tool triggers
            another human-interaction pause.
    """
    messages: List[Dict[str, Any]] = list(state.get("messages", []))
    tool_call_id: Optional[str] = state.get("tool_call_id")
    resolved_model = self._model_value(state.get("model") or state.get("agent_name"))
    turn_id = str(uuid.uuid4())

    # Inject the resumption value as a tool result or a new user turn.
    if tool_call_id:
        messages.append(
            {
                "role": "tool",
                "tool_call_id": tool_call_id,
                "name": "handoff_tool",
                "content": user_input,
            }
        )
    else:
        messages.append({"role": "user", "content": user_input})

    request_args: Dict[str, Any] = {
        "model": resolved_model,
        "messages": messages,
        "max_tokens": 4096,
        "temperature": 0.1,
        "stream": False,
    }
    if self.enable_tools:
        request_args["tools"] = self._prepare_zai_tools()
        request_args["tool_choice"] = "auto"

    response = await self._chat_completion(**request_args, use_tools=self.enable_tools)
    all_tool_calls: List[ToolCall] = []
    result = response.choices[0].message
    max_turns = 10
    turns = 0

    while getattr(result, "tool_calls", None) and turns < max_turns:
        turns += 1
        messages.append(self._message_to_dict(result))
        for provider_tc in result.tool_calls:
            fn = provider_tc.function
            tool_name = fn.name
            tool_args = self._parse_tool_arguments(fn.arguments)
            tc = ToolCall(
                id=provider_tc.id,
                name=tool_name,
                arguments=tool_args,
            )
            try:
                started = time.perf_counter()
                tool_result = await self._execute_tool(tool_name, tool_args)
                tc.execution_time = time.perf_counter() - started
                tc.result = tool_result
                content = json.dumps(tool_result, default=str)
            except Exception as exc:
                from parrot.core.exceptions import HumanInteractionInterrupt

                if isinstance(exc, HumanInteractionInterrupt):
                    exc.session_id = session_id
                    exc.messages = messages.copy()
                    exc.tool_call_id = provider_tc.id
                    exc.agent_name = resolved_model
                    raise
                tc.error = str(exc)
                content = f"Error: {exc}"
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": provider_tc.id,
                    "name": tool_name,
                    "content": content,
                }
            )
            all_tool_calls.append(tc)

        follow_up = dict(request_args)
        follow_up["messages"] = messages
        response = await self._chat_completion(**follow_up, use_tools=True)
        result = response.choices[0].message

    return self._create_ai_message(
        response=response,
        input_text="[Resumed Conversation]",
        model=resolved_model,
        user_id=state.get("user_id"),
        session_id=session_id,
        turn_id=turn_id,
        tool_calls=all_tool_calls,
    )

invoke async

invoke(prompt: str, *, output_type: Optional[type] = None, structured_output: Optional[StructuredOutputConfig] = None, model: Optional[str] = None, system_prompt: Optional[str] = None, max_tokens: Optional[int] = None, temperature: float = 0.0, use_tools: bool = False, tools: Optional[list] = None) -> InvokeResult

Lightweight stateless invocation for ZaiClient.

Makes a single chat.completions.create call without conversation history, retries, or the full prompt-builder overhead. Uses Z.ai's native json_schema response format for structured output.

PARAMETER	DESCRIPTION
prompt	User prompt. TYPE: str
output_type	Pydantic model or dataclass to parse the response into. Mutually exclusive with structured_output (the latter wins). TYPE: Optional[type] DEFAULT: None
structured_output	Full :class:StructuredOutputConfig. Takes precedence over output_type. TYPE: Optional[StructuredOutputConfig] DEFAULT: None
model	Model override. Falls back to an explicitly selected :attr:model, then :attr:_lightweight_model. TYPE: Optional[str] DEFAULT: None
system_prompt	System prompt override. Falls back to the default :attr:BASIC_SYSTEM_PROMPT template. TYPE: Optional[str] DEFAULT: None
max_tokens	Maximum completion tokens (default 4096). TYPE: Optional[int] DEFAULT: None
temperature	Sampling temperature (default 0.0 for deterministic structured extraction). TYPE: float DEFAULT: 0.0
use_tools	If True, inject registered tools into the request. TYPE: bool DEFAULT: False
tools	Additional tool definitions to register for this call. TYPE: Optional[list] DEFAULT: None

RETURNS	DESCRIPTION
InvokeResult	class:InvokeResult with output, model, usage, and
InvokeResult	raw_response.

RAISES	DESCRIPTION
	class:~parrot.exceptions.InvokeError: On any provider error.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def invoke(
    self,
    prompt: str,
    *,
    output_type: Optional[type] = None,
    structured_output: Optional[StructuredOutputConfig] = None,
    model: Optional[str] = None,
    system_prompt: Optional[str] = None,
    max_tokens: Optional[int] = None,
    temperature: float = 0.0,
    use_tools: bool = False,
    tools: Optional[list] = None,
) -> InvokeResult:
    """Lightweight stateless invocation for ZaiClient.

    Makes a single ``chat.completions.create`` call without conversation
    history, retries, or the full prompt-builder overhead.  Uses Z.ai's
    native ``json_schema`` response format for structured output.

    Args:
        prompt: User prompt.
        output_type: Pydantic model or dataclass to parse the response
            into.  Mutually exclusive with *structured_output* (the latter
            wins).
        structured_output: Full :class:`StructuredOutputConfig`.  Takes
            precedence over *output_type*.
        model: Model override.  Falls back to an explicitly selected
            :attr:`model`, then :attr:`_lightweight_model`.
        system_prompt: System prompt override.  Falls back to the default
            :attr:`BASIC_SYSTEM_PROMPT` template.
        max_tokens: Maximum completion tokens (default ``4096``).
        temperature: Sampling temperature (default ``0.0`` for
            deterministic structured extraction).
        use_tools: If ``True``, inject registered tools into the request.
        tools: Additional tool definitions to register for this call.

    Returns:
        :class:`InvokeResult` with ``output``, ``model``, ``usage``, and
        ``raw_response``.

    Raises:
        :class:`~parrot.exceptions.InvokeError`: On any provider error.
    """
    try:
        resolved_system = self._resolve_invoke_system_prompt(system_prompt)
        config = self._build_invoke_structured_config(output_type, structured_output)
        resolved_model = self._resolve_invoke_model(model)
        max_tokens = self._resolve_max_tokens(max_tokens, resolved_model, for_invoke=True)

        if tools:
            for tool_def in tools:
                self.register_tool(tool_def)

        messages: List[Dict[str, Any]] = [
            {"role": "system", "content": resolved_system},
            {"role": "user", "content": prompt},
        ]

        kwargs: Dict[str, Any] = {
            "model": resolved_model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": False,
        }

        if config:
            kwargs.update(
                self._prepare_structured_output_format(config.output_type)
                if config.format == OutputFormat.JSON
                else {}
            )

        if use_tools:
            tool_defs = self._prepare_zai_tools()
            if tool_defs:
                kwargs["tools"] = tool_defs
                kwargs["tool_choice"] = "auto"

        response = await self._chat_completion(**kwargs, use_tools=use_tools)
        raw_text = getattr(response.choices[0].message, "content", None) or ""

        output: Any = raw_text
        if config:
            # Known-truncated output must not reach a custom parser either.
            self._raise_if_truncated(self._extract_finish_reason(response), model=resolved_model)
            if config.custom_parser:
                output = config.custom_parser(raw_text)
            else:
                output = await self._parse_structured_output(
                    raw_text,
                    config,
                    finish_reason=self._extract_finish_reason(response),
                    model=resolved_model,
                )

        usage = self._usage_from_response(response)
        return self._build_invoke_result(output, output_type, resolved_model, usage, response)

    except InvokeError:
        raise
    except Exception as exc:
        raise self._handle_invoke_error(exc) from exc

embed async

embed(*args: Any, **kwargs: Any) -> Any

Embeddings are not implemented by this chat client yet.

Source code in packages/ai-parrot/src/parrot/clients/zai.py

async def embed(self, *args: Any, **kwargs: Any) -> Any:
    """Embeddings are not implemented by this chat client yet."""
    raise NotImplementedError("ZaiClient embed() is not implemented.")