picoware.system.agent.mcp.MCPClient

class documentation

class MCPClient:

Constructor: MCPClient(http, servers)

Client for standard MCP JSON-RPC requests over HTTP.

Method	`__init__`	Initialize an MCP client from the configured endpoints.
Method	`call_tool`	Call a discovered tool on the selected MCP server.
Method	`list_servers`	Return configured servers without exposing their API keys.
Method	`reset`	Reset the selected server and MCP session for a new agent run.
Method	`select_server`	Initialize a server and return its public details and MCP tools.
Class Variable	`__slots__`	Undocumented
Property	`enabled`	Return whether at least one MCP endpoint is configured.
Static Method	`_append_sse_event`	Undocumented
Static Method	`_decode_response`	Undocumented
Static Method	`_header`	Undocumented
Static Method	`_normalize_servers`	Undocumented
Static Method	`_public_server`	Undocumented
Method	`_find_server`	Undocumented
Method	`_headers`	Undocumented
Method	`_list_tools`	Undocumented
Method	`_request`	Undocumented
Instance Variable	`_http`	Undocumented
Instance Variable	`_next_id`	Undocumented
Instance Variable	`_protocol_version`	Undocumented
Instance Variable	`_selected_server`	Undocumented
Instance Variable	`_servers`	Undocumented
Instance Variable	`_session_id`	Undocumented

def __init__(self, http, servers: list[dict] | None = None): ¶

Initialize an MCP client from the configured endpoints.

Parameters
http:`HTTP`	The Picoware HTTP client.
servers:`list[dict] or None`	Configured MCP endpoint entries.

def call_tool(self, server_id: str, tool_name: str, arguments: dict) -> str: ¶

Call a discovered tool on the selected MCP server.

Parameters
server_id:`str`	The selected server id.
tool_name:`str`	The original MCP tool name.
arguments:`dict`	Tool arguments from the model.

Returns
`str`	The JSON-encoded MCP tool result.

def list_servers(self) -> list[dict]: ¶

Return configured servers without exposing their API keys.

def reset(self): ¶

Reset the selected server and MCP session for a new agent run.

def select_server(self, server_id: str) -> tuple[dict, list[dict]]: ¶

Initialize a server and return its public details and MCP tools.

Parameters
server_id:`str`	The id of the configured server to select.

Returns
`tuple[dict, list[dict]]`	Public server details and raw MCP tool schemas.

__slots__: list[str] = ¶

Undocumented

@property

enabled: bool = ¶

Return whether at least one MCP endpoint is configured.

@staticmethod

def _append_sse_event(events: list[dict], data_lines: list[str]): ¶

Undocumented

@staticmethod

def _decode_response(response, expected_id: int | None) -> dict | None: ¶

Undocumented

@staticmethod

def _header(headers, name: str): ¶

Undocumented

@staticmethod

def _normalize_servers(servers: list[dict] | None) -> list[dict]: ¶

Undocumented

@staticmethod

def _public_server(server: dict) -> dict: ¶

Undocumented

def _find_server(self, server_id: str) -> dict | None: ¶

Undocumented

def _headers(self, server: dict, include_protocol: bool = True) -> dict: ¶

Undocumented

def _list_tools(self, server: dict) -> list[dict]: ¶

Undocumented

def _request(self, server: dict, method: str, params: dict | None = None, include_id: bool = True) -> tuple[dict | None, object]: ¶

Undocumented

_http = ¶

Undocumented

_next_id: int = ¶

Undocumented

_protocol_version = ¶

Undocumented

_selected_server = ¶

Undocumented

_servers = ¶

Undocumented

_session_id = ¶

Undocumented