mindroot.coreplugins.mcp_ package

Submodules

mindroot.coreplugins.mcp_.catalog_commands module

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_add_custom(server_info: Dict[str, Any], context=None)

Add a custom server to the catalog

Example: { “mcp_catalog_add_custom”: {

“server_info”: {

“name”: “my_server”, “display_name”: “My Custom Server”, “description”: “Custom MCP server”, “command”: “python”, “args”: [“-m”, “my_mcp_server”], “category”: “custom”, “install_method”: “manual”

}

}}

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_categories(context=None)

Get list of server categories

Example: { “mcp_catalog_categories”: {} }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_info(server_name: str, context=None)

Get detailed info about a catalog server

Example: { “mcp_catalog_info”: { “server_name”: “calculator” } }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_install(server_name: str, context=None)

Install an MCP server from catalog

Example: { “mcp_catalog_install”: { “server_name”: “calculator” } }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_install_and_run(server_name: str, context=None)

Install and run an MCP server from catalog

Example: { “mcp_catalog_install_and_run”: { “server_name”: “calculator” } }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_list(category: str = None, context=None)

List MCP servers from catalog

Example: { “mcp_catalog_list”: {} } { “mcp_catalog_list”: { “category”: “utilities” } }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_refresh(context=None)

Refresh catalog and update server status

Example: { “mcp_catalog_refresh”: {} }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_search(query: str, context=None)

Search MCP servers in catalog

Example: { “mcp_catalog_search”: { “query”: “calculator” } } { “mcp_catalog_search”: { “query”: “database” } }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_status(context=None)

Get status of all catalog servers

Example: { “mcp_catalog_status”: {} }

async mindroot.coreplugins.mcp_.catalog_commands.mcp_catalog_stop(server_name: str, context=None)

Stop an MCP server

Example: { “mcp_catalog_stop”: { “server_name”: “calculator” } }

mindroot.coreplugins.mcp_.catalog_manager module

class mindroot.coreplugins.mcp_.catalog_manager.MCPCatalogManager(plugin_dir: str = None, working_dir: str = None)

Bases: object

Manages MCP server catalog and process detection

add_custom_server(server_info: Dict[str, Any])

Add a custom server to the catalog

create_empty_catalog()

Create empty catalog structure

detect_running_servers() → Dict[str, bool]

Detect which MCP servers are currently running

ensure_catalog_exists()

Ensure catalog file exists, create from default if needed

get_categories() → List[str]

Get list of all categories

get_running_processes() → List[Dict[str, Any]]

Get list of running processes with details

get_server_info(server_name: str) → Dict[str, Any] | None

Get detailed info for a specific server

get_servers_by_category(category: str = None) → Dict[str, Any]

Get servers filtered by category

init_catalog_from_default()

Initialize catalog from default servers

load_catalog() → Dict[str, Any]

Load the server catalog

mark_server_installed(server_name: str, installed: bool = True)

Mark a server as installed or not

remove_server(server_name: str)

Remove a server from the catalog

save_catalog(catalog: Dict[str, Any])

Save the server catalog

search_servers(query: str) → Dict[str, Any]

Search servers by name, description, or tags

update_server_status()

Update running status for all servers in catalog

mindroot.coreplugins.mcp_.dynamic_commands module

class mindroot.coreplugins.mcp_.dynamic_commands.MCPDynamicCommands

Bases: object

Handles dynamic registration of MCP tools as MindRoot commands

get_registered_commands() → List[str]

Get list of registered command names

async register_tools(server_name: str, tools: List[Any])

Register MCP tools as dynamic MindRoot commands

set_sessions(sessions: Dict[str, Any])

Set reference to MCP sessions

async unregister_server_tools(server_name: str)

Unregister all tools for a server

mindroot.coreplugins.mcp_.mcp_manager module

class mindroot.coreplugins.mcp_.mcp_manager.MCPManager

Bases: object

Manages MCP server connections and operations

add_server(name: str, server: MCPServer)

Add a new server configuration

async call_tool(server_name: str, tool_name: str, arguments: Dict[str, Any]) → Any

Call a tool on an MCP server

complete_oauth_flow(server_name: str, code: str, state: str | None = None) → bool

Complete OAuth flow with authorization code.

async connect_oauth_server(name: str) → bool

Connect to an OAuth-protected MCP server.

async connect_remote_server(name: str) → bool

Connect to a remote MCP server (HTTP/SSE).

async connect_server(name: str, secrets: Dict[str, str] | None = None) → bool

Connect to an MCP server

async disconnect_server(name: str) → bool

Disconnect from an MCP server

get_oauth_status(server_name: str) → Dict[str, Any]

Get OAuth flow status for a server.

async install_server(server_name: str) → bool

Install an MCP server

load_config()

Load server configurations from file

mark_server_as_installed(name: str, registry_id: str = None)

Mark server as installed (in-memory only for remote servers)

async read_resource(server_name: str, uri: str) → Any

Read a resource from an MCP server

remove_server(name: str)

Remove a server configuration

async sanity_test() → str

save_config()

Save server configurations to file - LOCAL SERVERS ONLY

start_oauth_flow(server_name: str) → str | None

Start OAuth flow and return authorization URL.

async test_local_server_capabilities(name: str, command: str, args: List[str], env: Dict[str, str], secrets: Dict[str, str] | None = None) → Dict[str, Any]

Test connection to a local MCP server and return its capabilities.

This method creates a temporary server, connects to it, extracts capabilities, and cleans up. It follows the same pattern as the working catalog system.

Parameters:

• name – Display name for the server (for error messages)

• command – Command to run the MCP server

• args – Arguments for the command

• env – Environment variables

• secrets – A dictionary of secrets to substitute into placeholders.

Returns:

Dict containing success status, message, and capabilities (tools, resources, prompts)

Raises:

Exception – If MCP SDK is not available or connection fails

class mindroot.coreplugins.mcp_.mcp_manager.MCPServer(*, name: str, description: str, command: str | None = None, args: List[str] = [], env: Dict[str, str] = {}, transport: str = 'stdio', url: str | None = None, provider_url: str | None = None, transport_url: str | None = None, transport_type: str | None = None, auth_type: str = 'none', auth_headers: Dict[str, str] = {}, authorization_server_url: str | None = None, client_id: str | None = None, client_secret: str | None = None, scopes: List[str] = [], redirect_uri: str | None = None, access_token: str | None = None, refresh_token: str | None = None, token_expires_at: str | None = None, status: str = 'disconnected', secrets: Dict[str, str] | None = None, capabilities: Dict[str, Any] = {}, install_method: str = 'manual', install_package: str | None = None, auto_install: bool = False, installed: bool = False)

Bases: BaseModel

Model for MCP server configuration

access_token: str | None

args: List[str]

auth_headers: Dict[str, str]

auth_type: str

authorization_server_url: str | None

auto_install: bool

capabilities: Dict[str, Any]

client_id: str | None

client_secret: str | None

command: str | None

description: str

env: Dict[str, str]

install_method: str

install_package: str | None

installed: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

provider_url: str | None

redirect_uri: str | None

refresh_token: str | None

scopes: List[str]

secrets: Dict[str, str] | None

status: str

token_expires_at: str | None

transport: str

transport_type: str | None

transport_url: str | None

url: str | None

mindroot.coreplugins.mcp_.mod module

async mindroot.coreplugins.mcp_.mod.enhanced_mcp_manager_service(context=None)

Service to access enhanced MCP manager (same as mcp_manager_service now)

async mindroot.coreplugins.mcp_.mod.handle_callback() → tuple[str, str | None]

async mindroot.coreplugins.mcp_.mod.handle_redirect(auth_url: str) → None

async mindroot.coreplugins.mcp_.mod.mcp_check_tools(context=None)

Check availability of installation tools

Example: { “mcp_check_tools”: {} }

async mindroot.coreplugins.mcp_.mod.mcp_connect(server_name: str, context=None)

Connect to an MCP server

Example: { “mcp_connect”: { “server_name”: “filesystem” } }

async mindroot.coreplugins.mcp_.mod.mcp_debug_connection(server_name: str, context=None)

Debug MCP server connection and dynamic command registration

Example: { “mcp_debug_connection”: { “server_name”: “calculator” } } { “mcp_debug_connection”: { “server_name”: “github” } }

async mindroot.coreplugins.mcp_.mod.mcp_disconnect(server_name: str, context=None)

Disconnect from an MCP server

Example: { “mcp_disconnect”: { “server_name”: “filesystem” } }

async mindroot.coreplugins.mcp_.mod.mcp_enhanced_connect(server_name: str, context=None)

Connect to MCP server with enhanced features

Example: { “mcp_enhanced_connect”: { “server_name”: “calculator” } }

async mindroot.coreplugins.mcp_.mod.mcp_enhanced_disconnect(server_name: str, context=None)

Disconnect from MCP server

Example: { “mcp_enhanced_disconnect”: { “server_name”: “calculator” } }

async mindroot.coreplugins.mcp_.mod.mcp_install_npx_server(name: str, package: str, description: str = None, context=None)

Install and configure an npx-based MCP server

Example: { “mcp_install_npx_server”: {

“name”: “github”, “package”: “@modelcontextprotocol/server-github”, “description”: “GitHub server”

} }

async mindroot.coreplugins.mcp_.mod.mcp_install_uvx_server(name: str, package: str, description: str = None, context=None)

Install and configure a uvx-based MCP server

Example: { “mcp_install_uvx_server”: {

“name”: “calculator”, “package”: “mcp-server-calculator”, “description”: “Calculator server”

} }

async mindroot.coreplugins.mcp_.mod.mcp_list_dynamic_commands(context=None)

List dynamically registered MCP commands

Example: { “mcp_list_dynamic_commands”: {} }

async mindroot.coreplugins.mcp_.mod.mcp_list_servers(context=None)

List all configured MCP servers

Example: { “mcp_list_servers”: {} }

async mindroot.coreplugins.mcp_.mod.mcp_manager_service(context=None)

Service to access the MCP manager

async mindroot.coreplugins.mcp_.mod.mcp_refresh_dynamic_commands(context=None)

Refresh dynamic command registration for all connected servers

Example: { “mcp_refresh_dynamic_commands”: {} }

mindroot.coreplugins.mcp_.oauth_storage module

OAuth token storage implementation for MCP servers.

class mindroot.coreplugins.mcp_.oauth_storage.MCPTokenStorage(server_name: str, mcp_manager)

Bases: TokenStorage

Token storage implementation for MCP OAuth flows.

Stores tokens and client information in the MCP server configuration.

clear_tokens()

Clear stored tokens for the server.

async get_client_info() → OAuthClientInformationFull | None

Get stored OAuth client information.

async get_tokens() → OAuthToken | None

Get stored OAuth tokens for the server.

async set_client_info(client_info: OAuthClientInformationFull) → None

Store OAuth client information.

async set_tokens(tokens: OAuthToken) → None

Store OAuth tokens for the server.

mindroot.coreplugins.mcp_.server_installer module

class mindroot.coreplugins.mcp_.server_installer.MCPServerInstaller

Bases: object

Handles installation of MCP servers via different methods

async static check_tools() → Dict[str, bool]

Check if installation tools are available

async static install_with_npm(package: str, args: list = None) → bool

Install package with npm

async static install_with_npx(package: str, args: list = None) → bool

Install package with npx (similar to uvx, runs without global install)

async static install_with_pip(package: str, args: list = None) → bool

Install package with pip

async static install_with_uvx(package: str, args: list = None) → bool

Install package with uvx

mindroot.coreplugins.mcp_.setup module

mindroot.coreplugins.mcp_.test_dynamic_commands module

mindroot.coreplugins.mcp_.testmcpclient module

Before running, specify running MCP RS server URL. To spin up RS server locally, see

examples/servers/simple-auth/README.md

cd to the examples/snippets directory and run:

uv run oauth-client

class mindroot.coreplugins.mcp_.testmcpclient.InMemoryTokenStorage

Bases: TokenStorage

Demo In-memory token storage implementation.

async get_client_info() → OAuthClientInformationFull | None

Get stored client information.

async get_tokens() → OAuthToken | None

Get stored tokens.

async set_client_info(client_info: OAuthClientInformationFull) → None

Store client information.

async set_tokens(tokens: OAuthToken) → None

Store tokens.

async mindroot.coreplugins.mcp_.testmcpclient.handle_callback() → tuple[str, str | None]

async mindroot.coreplugins.mcp_.testmcpclient.handle_redirect(auth_url: str) → None

async mindroot.coreplugins.mcp_.testmcpclient.main()

Run the OAuth client example.

mindroot.coreplugins.mcp_.testmcpclient.run()

Module contents