axio-tools-mcp¶
Explicit MCP server connections and conversion of discovered MCP tools into
Axio Tool definitions.
- class axio_tools_mcp.MCPServerConfig(name: str, command: str | None = None, args: list[str] = <factory>, env: dict[str, str] | None=None, url: str | None = None, headers: dict[str, str]=<factory>, timeout: float = 30.0, scope: str = 'global', protocol_version: str | None = None, cwd: str | None = None, inherit_env: bool = True, enabled: bool = True, env_from: dict[str, str]=<factory>, headers_from: dict[str, str]=<factory>, oauth: dict[str, str | int] | None=None)[source]¶
Describe one stdio or HTTP server using values from an MCP JSON entry.
from_dictconsumes an already parsed entry. It does not read files or discover roots. Applications choose names and resolve source-relative paths.connectconstructs and opens the corresponding session. The caller owns the returned session and must close it.- async connect(session_type: type[MCPSession] | None = None) MCPSession[source]¶
Create and open a session; close partial startup on every failure.
- classmethod from_dict(name: str, data: Mapping[str, Any]) MCPServerConfig[source]¶
Validate one decoded JSON server entry without performing I/O.
- classmethod from_settings(name: str, data: dict[str, str]) MCPServerConfig[source]¶
Deserialize from flat string dict.
- resolve(environment: Mapping[str, str] | None = None) MCPServerConfig[source]¶
Resolve references and executable for this connection attempt.
- class axio_tools_mcp.MCPSession(config: MCPServerConfig)[source]¶
Manages connection lifecycle for a single MCP server.
- async call_tool(name: str, arguments: dict[str, Any]) CallToolResult[source]¶
Call a tool on the connected MCP server.
- connect_http(adapter: Adapter) Client[source]¶
Open an HTTP session for the Streamable HTTP transport.
- async connect_stdio(adapter: Adapter) StdioClient[source]¶
Start the server process and talk to its standard streams.
- environment() dict[str, str][source]¶
Return configured variables, optionally added to the parent environment.
- exclusive() AbstractAsyncContextManager[Any][source]¶
Serialize requests on stdio; HTTP requests are independent.
- async handshake() None[source]¶
Initialize, then follow the server to the revision it answered with.
- async list_resource_templates() list[ResourceTemplateDef][source]¶
List every resource template using the connection’s pagination and lock.
- async list_resources() list[ResourceDef][source]¶
List resources using the same connection and stdio lock as tools.
- async log_stderr(stream: StreamReader) None[source]¶
Forward the server’s standard error to the logger as warnings.
- async open(protocol_version: str) None[source]¶
Start the transport and complete the handshake on one revision.
- property protocol_version: str | None¶
Return the revision in use, or None while disconnected.
- async axio_tools_mcp.load_mcp_tools(servers: list[MCPServerConfig]) tuple[list[Tool[Any]], list[MCPSession]][source]¶
Connect to MCP servers and discover their tools.
Returns
(tools, sessions). The caller is responsible for closing sessions. Failed servers are logged and skipped.
- class axio_tools_mcp.MCPRegistry(on_status: ~collections.abc.Callable[[str], None] = <function MCPRegistry.<lambda>>, _instructions: dict[str, tuple[~axio_tools_mcp.session.MCPSession, tuple[~axio.instructions.Instruction, ...]]] = <factory>, _configs: dict[str, ~axio_tools_mcp.config.MCPServerConfig] = <factory>, _sessions: dict[str, ~axio_tools_mcp.session.MCPSession] = <factory>, _tools: dict[str, list[~axio.tool.Tool[~typing.Any]]] = <factory>, _resources: dict[str, ~axio_tools_mcp.resources.ResourceCatalog] = <factory>, _skills: dict[str, ~axio_tools_mcp.skills.SkillCatalog] = <factory>, _errors: dict[str, str] = <factory>, _discovery_errors: dict[str, list[str]] = <factory>, _config_db: ~typing.Any = None, _global_config_db: ~typing.Any = None, _server_scope: dict[str, ~typing.Any] = <factory>)[source]¶
Manages MCP server instances, discovers tools, persists to config DB.
- async add_server(config: MCPServerConfig, scope: Any = None) list[Tool[Any]][source]¶
Add new server, connect, load tools, persist config.
- async init(config: Any = None, global_config: Any = None) None[source]¶
Load saved MCP server configs from both DBs, connect, discover tools.
- inspect_servers() list[ServerInspection][source]¶
Return current connection details and cached catalogs for each saved server.
- async reconnect() None[source]¶
Reconnect all saved servers without changing their configuration or scope.
- server_config(name: str) MCPServerConfig[source]¶
Return the config for a given server.
- property server_names: list[str]¶
Names of all configured servers.
- skill_catalogs() dict[str, SkillCatalog][source]¶
Return skill catalogs by server name for interactive browsers.
- async update_server(name: str, config: MCPServerConfig, scope: Any = None) list[Tool[Any]][source]¶
Reconnect server with new config.
Plugin¶
MCPPlugin is what the axio.tools.settings entry point resolves to, so a host
application can offer MCP server configuration without importing this package.