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(conversation_memory: Optional[ConversationMemory] = None, 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,
    conversation_memory: Optional[ConversationMemory] = None,
    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', 4096)
    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)
        self.max_tokens = kwargs.get('max_tokens', 4096)
    self.conversation_memory = conversation_memory or InMemoryConversation()
    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
    # Fallback model for capacity errors (subclasses set their own default)
    self._fallback_model: Optional[str] = kwargs.get('fallback_model', None)
    # 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

start_conversation async

start_conversation(user_id: str, session_id: str, metadata: Optional[Dict[str, Any]] = None, chatbot_id: Optional[str] = None) -> ConversationHistory

Start a new conversation session.

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

async def start_conversation(
    self,
    user_id: str,
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    chatbot_id: Optional[str] = None,
) -> ConversationHistory:
    """Start a new conversation session."""
    return await self.conversation_memory.create_history(
        user_id,
        session_id,
        metadata=metadata,
        chatbot_id=self._get_chatbot_key(chatbot_id)
    )

get_conversation async

get_conversation(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[ConversationHistory]

Get an existing conversation session.

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

async def get_conversation(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get an existing conversation session."""
    if not self.conversation_memory:
        return None
    return await self.conversation_memory.get_history(
        user_id,
        session_id,
        chatbot_id=self._get_chatbot_key(chatbot_id)
    )

clear_conversation async

clear_conversation(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Clear conversation history for a session.

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

async def clear_conversation(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Clear conversation history for a session."""
    if not self.conversation_memory:
        return False
    await self.conversation_memory.clear_history(
        user_id,
        session_id,
        chatbot_id=self._get_chatbot_key(chatbot_id)
    )
    return True

delete_conversation async

delete_conversation(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Delete conversation history entirely.

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

async def delete_conversation(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Delete conversation history entirely."""
    if not self.conversation_memory:
        return False
    return await self.conversation_memory.delete_history(
        user_id,
        session_id,
        chatbot_id=self._get_chatbot_key(chatbot_id)
    )

list_user_conversations async

list_user_conversations(user_id: str, chatbot_id: Optional[str] = None) -> List[str]

List all conversation sessions for a user.

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

async def list_user_conversations(
    self,
    user_id: str,
    chatbot_id: Optional[str] = None
) -> List[str]:
    """List all conversation sessions for a user."""
    if not self.conversation_memory:
        return []
    return await self.conversation_memory.list_sessions(
        user_id,
        chatbot_id=self._get_chatbot_key(chatbot_id)
    )

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, plt_style: str = 'seaborn-v0_8-whitegrid', palette: str = 'Set2') -> 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
plt_style	Matplotlib style TYPE: str DEFAULT: 'seaborn-v0_8-whitegrid'
palette	Seaborn color palette TYPE: str DEFAULT: 'Set2'

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,
    plt_style: str = 'seaborn-v0_8-whitegrid',
    palette: str = 'Set2'
) -> PythonREPLTool:
    """Register Python REPL tool with a ClaudeAPIClient.

    Args:
        client: The ClaudeAPIClient instance
        report_dir: Directory for saving reports
        plt_style: Matplotlib style
        palette: Seaborn color palette

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

    tool = PythonREPLTool(
        report_dir=report_dir,
        plt_style=plt_style,
        palette=palette,
        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: int = 4096, temperature: float = 0.7, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[str] = None, structured_output: Union[type, StructuredOutputConfig, None] = None, user_id: Optional[str] = None, session_id: Optional[str] = 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 TYPE: int DEFAULT: 4096
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
structured_output	Optional structured output configuration TYPE: Union[type, StructuredOutputConfig, None] DEFAULT: None
user_id	Optional user identifier for tracking TYPE: Optional[str] DEFAULT: None
session_id	Optional session identifier for tracking TYPE: Optional[str] 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: int = 4096,
    temperature: float = 0.7,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[str] = None,
    structured_output: Union[type, StructuredOutputConfig, None] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = 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
        temperature: Sampling temperature for response generation
        files: Optional files to include in the request
        system_prompt: Optional system prompt to guide the model
        structured_output: Optional structured output configuration
        user_id: Optional user identifier for tracking
        session_id: Optional session identifier for tracking
        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: int = 4096, temperature: float = 0.7, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[str] = None, user_id: Optional[str] = None, session_id: Optional[str] = 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: int = 4096,
    temperature: float = 0.7,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[str] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = 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: int = 4096, 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 _lightweight_model, then self.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 (default 4096). TYPE: int DEFAULT: 4096
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: int = 4096,
    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
            ``_lightweight_model``, then ``self.model``.
        system_prompt: Override the system prompt. Falls back to
            ``BASIC_SYSTEM_PROMPT`` rendered with instance attributes.
        max_tokens: Maximum completion tokens (default 4096).
        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().")

create_conversation_memory staticmethod

create_conversation_memory(memory_type: str = 'memory', **kwargs) -> ConversationMemory

Factory method to create a conversation memory instance.

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

@staticmethod
def create_conversation_memory(
    memory_type: str = "memory",
    **kwargs
) -> ConversationMemory:
    """Factory method to create a conversation memory instance."""
    if memory_type == "memory":
        return InMemoryConversation()
    elif memory_type == "redis":
        return RedisConversation(**kwargs)
    elif memory_type == "file":
        return FileConversationMemory(**kwargs)
    else:
        raise ValueError(
            f"Unsupported memory type: {memory_type}"
        )

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

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: AbstractClient

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

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:
    self.api_key = api_key or config.get("ZAI_API_KEY")
    if not self.api_key:
        raise ValueError(
            "ZAI_API_KEY is required. Pass api_key= or set the ZAI_API_KEY environment variable."
        )
    self.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
    self.base_headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {self.api_key}",
    }
    super().__init__(**kwargs)

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: int = 4096, temperature: float = 0.7, top_p: float = 0.9, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[Union[str, list]] = None, structured_output: Union[type, StructuredOutputConfig, None] = None, user_id: Optional[str] = None, session_id: Optional[str] = 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: int DEFAULT: 4096
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. TYPE: Optional[str] DEFAULT: None
session_id	Optional session identifier for conversation memory. TYPE: Optional[str] DEFAULT: None
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: int = 4096,
    temperature: float = 0.7,
    top_p: float = 0.9,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[Union[str, list]] = None,
    structured_output: Union[type, StructuredOutputConfig, None] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = 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``.
    """
    resolved_model = self._model_value(model)
    turn_id = str(uuid.uuid4())
    started = time.perf_counter()
    messages, conversation_history, resolved_system_prompt = await self._build_messages(
        prompt,
        files,
        user_id,
        session_id,
        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._create_completion(**request_args)
        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)

        response_time = time.perf_counter() - started
        ai_message = self._create_ai_message(
            response=response,
            input_text=prompt,
            model=resolved_model,
            user_id=user_id,
            session_id=session_id,
            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,
    )
    await self._update_conversation_memory(
        user_id,
        session_id,
        conversation_history,
        messages,
        resolved_system_prompt,
        turn_id,
        prompt,
        ai_message.response or "",
        tools_used=[tool_call.name for tool_call in all_tool_calls],
    )
    return ai_message

ask_stream async

ask_stream(prompt: str, model: Union[str, ZaiModel, None] = None, max_tokens: int = 4096, temperature: float = 0.7, top_p: float = 0.9, files: Optional[List[Union[str, Path]]] = None, system_prompt: Optional[Union[str, list]] = None, user_id: Optional[str] = None, session_id: Optional[str] = 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: int DEFAULT: 4096
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. TYPE: Optional[str] DEFAULT: None
session_id	Optional session identifier for conversation memory. TYPE: Optional[str] DEFAULT: None
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: int = 4096,
    temperature: float = 0.7,
    top_p: float = 0.9,
    files: Optional[List[Union[str, Path]]] = None,
    system_prompt: Optional[Union[str, list]] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = 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``.
    """
    resolved_model = self._model_value(model)
    turn_id = str(uuid.uuid4())
    started = time.perf_counter()
    messages, conversation_history, resolved_system_prompt = await self._build_messages(
        prompt,
        files,
        user_id,
        session_id,
        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 self._stream_completion(**request_args):
            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 self._stream_completion(**follow_up_args):
                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=user_id,
            session_id=session_id,
            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,
    )
    await self._update_conversation_memory(
        user_id,
        session_id,
        conversation_history,
        messages,
        resolved_system_prompt,
        turn_id,
        prompt,
        content_text,
        tools_used=[tool_call.name for tool_call in all_tool_calls],
    )
    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._create_completion(**request_args)
    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._create_completion(**follow_up)
        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: int = 4096, 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 :attr:_lightweight_model, then :attr: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: int DEFAULT: 4096
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: int = 4096,
    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 :attr:`_lightweight_model`,
            then :attr:`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)

        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._create_completion(**kwargs)
        raw_text = getattr(response.choices[0].message, "content", None) or ""

        output: Any = raw_text
        if config:
            if config.custom_parser:
                output = config.custom_parser(raw_text)
            else:
                output = await self._parse_structured_output(raw_text, config)

        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.")