Client and parameter API¶

Start with tool execution for orchestration behavior and web applications for resource ownership and per-request context. Constructor configuration is shared for the client lifespan; request identity belongs in trusted context.

Async and blocking clients¶

`seqoria_agent.client.ai_client.async_client.AsyncAIClient` ¶

MCP-native AI client for tool-augmented LLM interactions (async version).

This is the asynchronous version of the AI client. For synchronous usage, use :class:AIClient instead.

Examples:

>>> async with AsyncAIClient(providers=[AsyncOpenAIProvider()]) as client:
...     async for chunk in client.stream_chat("Hello!", history=[]):
...         print(chunk)

`add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

Convenience wrapper for adding a single MCP server.

`add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

Add MCP servers after initialization.

If connect=True and the client is already started (inside async with), this will connect immediately.
If refresh_tools=True, the ToolRegistry is refreshed so newly available tools become visible via list_tools() and to the LLM.

Returns the number of new server configs added (deduped by URL).

`block_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

Add tool names to the blocklist and optionally refresh the registry.

`call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any]` `async` ¶

Call an MCP tool directly.

`chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None]` `async` ¶

Chat with the LLM, handling tool calls automatically.

`close() -> None` `async` ¶

Close all connections and cleanup resources.

`list_mcp_servers() -> List[MCPServerConfig]` ¶

List configured MCP servers (read-only copy).

`list_tools() -> List[dict]` ¶

List currently available tools (OpenAI-compatible tool schema).

Notes

This reflects the most recent ToolRegistry.refresh() results.
For MCP-backed tools, you typically want to call this after entering the client context (async with client:) or after calling await client.refresh_tools().

`llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList]` `async` ¶

Single LLM generation step.

`llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None]` `async` ¶

Stream a single LLM generation step.

`refresh_tools() -> None` `async` ¶

Manually refresh the tool registry.

`remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

Convenience wrapper for removing a single MCP server.

`remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

Remove MCP servers after initialization.

If disconnect=True, active connections are closed and routing entries for their sessions are pruned.
If refresh_tools=True, ToolRegistry is refreshed so removed tools stop being exposed to the LLM and list_tools().

Returns number of server configs removed (deduped by URL).

`run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> AsyncGenerator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None]` ¶

Stream tool call results as they complete.

`run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError]` `async` ¶

Execute tool calls and return results.

`set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

Replace the blocked-tool set and optionally refresh the registry.

`stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolMessages], None]` `async` ¶

Stream a chat session with the LLM.

`tools_enabled() -> bool` ¶

Return True if any tools are available.

`unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

Remove tool names from the blocklist and optionally refresh the registry.

`seqoria_agent.client.ai_client.sync_client.AIClient` ¶

MCP-native AI client for tool-augmented LLM interactions (sync version).

This is the synchronous version of the AI client. It wraps :class:AsyncAIClient and provides blocking methods for all operations.

For async/await usage, use :class:AsyncAIClient instead.

Examples:

>>> with AIClient(providers=[OpenAIProvider()]) as client:
...     response = client.chat_sync("Hello!", history=[])
...     print(response)

>>> with AIClient(providers=[OpenAIProvider()]) as client:
...     for chunk in client.stream_chat("Hello!", history=[]):
...         print(chunk.content, end="")

`async_client: AsyncAIClient` `property` ¶

Access the underlying AsyncAIClient for advanced async usage.

`executor` `property` ¶

Access the tool executor.

`llm` `property` ¶

Access the LLM interface.

`pool` `property` ¶

Access the MCP connection pool.

`registry` `property` ¶

Access the tool registry.

`add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int` ¶

Add a single MCP server after initialization (blocking).

`add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int` ¶

Add MCP servers after initialization (blocking).

`block_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

Add tool names to the blocklist and optionally refresh the tool registry.

`call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any]` ¶

Call an MCP tool directly (blocking).

`chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None, None]` ¶

Chat with the LLM, handling tool calls automatically (blocking generator).

This method yields responses as they are generated, handling tool calls automatically in the background.

`chat_sync(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AssistantMessage` ¶

Chat with the LLM and return final response (blocking, non-streaming).

This is a convenience method that consumes the entire chat loop and returns only the final assistant message.

`close() -> None` ¶

Close all connections and cleanup resources.

`list_mcp_servers() -> List[MCPServerConfig]` ¶

List configured MCP servers (read-only copy).

`list_tools() -> List[dict]` ¶

List currently available tools (OpenAI-compatible tool schema).

`llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList]` ¶

Single LLM generation step (blocking).

`llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None, None]` ¶

Stream a single LLM generation step (blocking generator).

`refresh_tools() -> None` ¶

Manually refresh the tool registry (blocking).

`remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int` ¶

Remove a single MCP server after initialization (blocking).

`remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int` ¶

Remove MCP servers after initialization (blocking).

`run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> Generator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None, None]` ¶

Stream tool call results as they complete (blocking generator).

`run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError]` ¶

Execute tool calls and return results (blocking).

`set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

Replace the blocked-tool set and optionally refresh the tool registry.

`stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolMessages], None, None]` ¶

Stream a chat session with the LLM (blocking generator).

Yields chunks as they are received, including tool call results.

`tools_enabled() -> bool` ¶

Return True if any tools are available.

`unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

Remove tool names from the blocklist and optionally refresh the tool registry.

Settings¶

`seqoria_agent.client.settings.ClientSettings` `dataclass` ¶

`copy() -> 'ClientSettings'` ¶

Copy configuration containers while retaining provider/resource ownership.

`with_provider(provider: Union[AsyncBaseProvider, BaseProvider]) -> 'ClientSettings'` ¶

Return a copy with provider appended to :attr:providers.

Hidden local parameters¶

ParameterRegistry is the single registry for hidden local arguments. Request context takes precedence. Sources can be literals or sync/async callables receiving matching context keys. A missing value or source failure prevents the affected tool from running. Direct LocalToolbox.call_tool() callers supply trusted server_args explicitly.

`seqoria_agent.client.parameters.ParameterRegistry` ¶

Resolve server-populated parameters from a mapping.

Examples:

async def demo(): async def get_user_id(tenant_id: str | None = None) -> str: return "u-42" if tenant_id == "acme" else "anonymous"

reg = ParameterRegistry({
    "tenant_id": "acme",
    "user_id": get_user_id,
})
ctx: ParamContext = {"tenant_id": "acme"}
assert await reg.resolve("tenant_id", context=ctx) == "acme"
assert await reg.resolve("user_id", context=ctx) == "u-42"
assert await reg.resolve("missing", context=ctx) is None

`init(mapping: Mapping[ParameterKey, ParameterSource])` ¶

Create a new registry from mapping.

Each key can be either a string parameter name or a subclass of :class:pydantic.BaseModel. For BaseModel keys the corresponding parameter name uses a lower-case first letter (AddressInfo → "addressInfo"). This makes it possible to group several related fields into a single structured object passed to the tool.

`resolve(name: str, *, context: ParamContext) -> Optional[Any]` `async` ¶

Return the resolved value for name or None when unresolved.

Resolution order: 1. A value in trusted request context takes precedence. 2. If name is absent from the registry mapping → None. 3. If the source is not callable → return it as-is. 4. If the source is callable → inspect its signature and pass the subset of context kwargs matching the callable parameters. Both sync and async callables are supported. Exceptions are caught and logged, the method returns None.

Client and parameter API¶

Async and blocking clients¶

seqoria_agent.client.ai_client.async_client.AsyncAIClient ¶

add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int async ¶

add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int async ¶

block_tools(names: Iterable[str], *, refresh: bool = True) -> None async ¶

call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any] async ¶

chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None] async ¶

close() -> None async ¶

list_mcp_servers() -> List[MCPServerConfig] ¶

list_tools() -> List[dict] ¶

llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList] async ¶

llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None] async ¶

refresh_tools() -> None async ¶

remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int async ¶

remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int async ¶

run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> AsyncGenerator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None] ¶

run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError] async ¶

set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None async ¶

stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolMessages], None] async ¶

tools_enabled() -> bool ¶

unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None async ¶

seqoria_agent.client.ai_client.sync_client.AIClient ¶

async_client: AsyncAIClient property ¶

executor property ¶

llm property ¶

pool property ¶

registry property ¶

add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int ¶

add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int ¶

block_tools(names: Iterable[str], *, refresh: bool = True) -> None ¶

call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any] ¶

chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None, None] ¶

chat_sync(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AssistantMessage ¶

close() -> None ¶

list_mcp_servers() -> List[MCPServerConfig] ¶

list_tools() -> List[dict] ¶

llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList] ¶

llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None, None] ¶

refresh_tools() -> None ¶

remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int ¶

remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int ¶

run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> Generator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None, None] ¶

run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError] ¶

set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None ¶

stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolMessages], None, None] ¶

tools_enabled() -> bool ¶

unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None ¶

Settings¶

seqoria_agent.client.settings.ClientSettings dataclass ¶

copy() -> 'ClientSettings' ¶

with_provider(provider: Union[AsyncBaseProvider, BaseProvider]) -> 'ClientSettings' ¶

Hidden local parameters¶

seqoria_agent.client.parameters.ParameterRegistry ¶

__init__(mapping: Mapping[ParameterKey, ParameterSource]) ¶

resolve(name: str, *, context: ParamContext) -> Optional[Any] async ¶

`seqoria_agent.client.ai_client.async_client.AsyncAIClient` ¶

`add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

`add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

`block_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

`call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any]` `async` ¶

`chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None]` `async` ¶

`close() -> None` `async` ¶

`list_mcp_servers() -> List[MCPServerConfig]` ¶

`list_tools() -> List[dict]` ¶

`llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList]` `async` ¶

`llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None]` `async` ¶

`refresh_tools() -> None` `async` ¶

`remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

`remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int` `async` ¶

`run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> AsyncGenerator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None]` ¶

`run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError]` `async` ¶

`set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

`stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AsyncGenerator[Union[ChatChunk, AssistantMessage, ToolMessages], None]` `async` ¶

`tools_enabled() -> bool` ¶

`unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None` `async` ¶

`seqoria_agent.client.ai_client.sync_client.AIClient` ¶

`async_client: AsyncAIClient` `property` ¶

`executor` `property` ¶

`llm` `property` ¶

`pool` `property` ¶

`registry` `property` ¶

`add_mcp_server(server: str | MCPServerConfig, *, connect: bool = True, refresh_tools: bool = True) -> int` ¶

`add_mcp_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True, refresh_tools: bool = True) -> int` ¶

`block_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

`call_mcp_tool(name: str, arguments: dict[str, Any], *, server_args: dict[str, Any] | None = None) -> List[Any]` ¶

`chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[AssistantMessage, ToolCallRequestList, ToolMessages], None, None]` ¶

`chat_sync(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> AssistantMessage` ¶

`close() -> None` ¶

`list_mcp_servers() -> List[MCPServerConfig]` ¶

`list_tools() -> List[dict]` ¶

`llm_generate_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Union[AssistantMessage, ToolCallRequestList]` ¶

`llm_stream_once(history: HistoryInput, *, append_user_message: Union[UserContentInput, None] = None, use_tools: bool = True, tool_choice: str = 'auto', model: str | None = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolCallRequestList], None, None]` ¶

`refresh_tools() -> None` ¶

`remove_mcp_server(server: str | MCPServerConfig, *, disconnect: bool = True, refresh_tools: bool = True) -> int` ¶

`remove_mcp_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True, refresh_tools: bool = True) -> int` ¶

`run_call_stream(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> Generator[Tuple[ToolCallRequest, ToolCallResult | ToolCallError], None, None]` ¶

`run_tool_calls(tool_calls: List[ToolCallRequest], *, context: ParamContext | None = None) -> List[ToolCallResult | ToolCallError]` ¶

`set_blocked_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

`stream_chat(user_query: str, history: HistoryInput, *, max_turns: int = 5, model: str | None = None, context: ParamContext | None = None, user_message_content: Union[UserContentInput, None] = None) -> Generator[Union[ChatChunk, AssistantMessage, ToolMessages], None, None]` ¶

`tools_enabled() -> bool` ¶

`unblock_tools(names: Iterable[str], *, refresh: bool = True) -> None` ¶

`seqoria_agent.client.settings.ClientSettings` `dataclass` ¶

`copy() -> 'ClientSettings'` ¶

`with_provider(provider: Union[AsyncBaseProvider, BaseProvider]) -> 'ClientSettings'` ¶

`seqoria_agent.client.parameters.ParameterRegistry` ¶

`init(mapping: Mapping[ParameterKey, ParameterSource])` ¶

`resolve(name: str, *, context: ParamContext) -> Optional[Any]` `async` ¶