Skip to main content

mcp.config_mcp_servers

MCP Server definitions and tool discovery.

This module reads MCP server configurations from a mcp.json file located at $HOME/.datalayer/mcp.json and discovers tools from them.

get_mcp_config_path​

def get_mcp_config_path() -> Path

Get the path to the MCP configuration file.

Returns:

Path to mcp.json file

expand_env_vars​

def expand_env_vars(value: str) -> str

Expand environment variables in a string.

Supports ${VAR_NAME} syntax.

Arguments:

  • value - String potentially containing env var references

Returns:

String with env vars expanded

expand_config_env_vars​

def expand_config_env_vars(config: dict[str, Any]) -> dict[str, Any]

Recursively expand environment variables in a config dictionary.

Arguments:

  • config - Configuration dictionary

Returns:

Config with env vars expanded

load_mcp_config​

def load_mcp_config() -> dict[str, Any]

Load MCP configuration from the mcp.json file.

Returns:

Dictionary containing mcpServers configuration

get_mcp_servers_from_config​

def get_mcp_servers_from_config() -> list[dict[str, Any]]

Get MCP server configurations from the config file.

Returns:

List of server configuration dictionaries

discover_mcp_server_tools​

async def discover_mcp_server_tools(
        server_config: dict[str, Any],
        timeout: float = 30.0) -> list[MCPServerTool]

Discover tools from an MCP server by starting it and listing tools.

Arguments:

  • server_config - MCP server configuration dictionary
  • timeout - Timeout in seconds for tool discovery

Returns:

List of MCPServerTool objects

create_mcp_servers_with_tools​

async def create_mcp_servers_with_tools(
        discover_tools: bool = True,
        tool_discovery_timeout: float = 30.0) -> list[MCPServer]

Create MCP server configurations with tool discovery.

Arguments:

  • discover_tools - Whether to discover tools from servers
  • tool_discovery_timeout - Timeout for tool discovery per server

Returns:

List of MCPServer objects with tools populated

get_config_mcp_servers​

async def get_config_mcp_servers(
        force_refresh: bool = False,
        discover_tools: bool = True) -> list[MCPServer]

Get the cached config MCP servers (from ~/.datalayer/mcp.json), initializing if needed.

Arguments:

  • force_refresh - Force re-initialization
  • discover_tools - Whether to discover tools from servers

Returns:

List of MCPServer objects

get_config_mcp_servers_sync​

def get_config_mcp_servers_sync() -> list[MCPServer]

Get cached config MCP servers synchronously.

Note: Returns empty list if not yet initialized. Use get_config_mcp_servers() in async context for proper initialization.

Returns:

List of MCPServer objects or empty list if not initialized

initialize_config_mcp_servers​

async def initialize_config_mcp_servers(
        discover_tools: bool = True) -> list[MCPServer]

Initialize config MCP servers (from ~/.datalayer/mcp.json) during application startup.

This should be called during FastAPI/Jupyter server startup.

Respects the AGENT_RUNTIMES_NO_CONFIG_MCP_SERVERS environment variable: if set to "true", returns an empty list instead of loading config servers.

Arguments:

  • discover_tools - Whether to discover tools from servers

Returns:

List of initialized MCPServer objects, or empty list if config servers are disabled