ResourceClient

ResourceClient

ResourceClient(
    port: int,
    host: str = "localhost",
    connect_retries: int = 10,
    connect_retry_interval: float = 1.0,
)

Context manager for

• loading the source code of Python modules and generated MCP client functions from an ExecutionContainer.
• generating Python client functions from MCP server tool metadata and storing the generated sources in an ExecutionContainer.

Parameters:

Name	Type	Description	Default
port	int	Host port for the container's resource port	required
host	str	Hostname or IP address of the container's host	'localhost'
connect_retries	int	Number of connection retries.	10
connect_retry_interval	float	Delay between connection retries in seconds.	1.0

Source code in ipybox/resource/client.py

def __init__(
    self,
    port: int,
    host: str = "localhost",
    connect_retries: int = 10,
    connect_retry_interval: float = 1.0,
):
    self.port = port
    self.host = host
    self._base_url = f"http://{self.host}:{self.port}"
    self._session: aiohttp.ClientSession = None
    self._connect_retries = connect_retries
    self._connect_retry_interval = connect_retry_interval

generate_mcp_sources async

generate_mcp_sources(
    relpath: str,
    server_name: str,
    server_params: dict[str, Any],
) -> list[str]

Generate Python client functions for tools provided by an MCP server.

One MCP client function is generated per MCP tool from its metadata. The generated function is stored in a module named /app/{relpath}/{server_name}/{tool_name}.py. Importing this module and calling the function invokes the corresponding MCP tool. This works for both stdio and sse based MCP servers. stdio based MCP servers are executed inside the container, sse based MCP servers are expected to run elsewhere.

Parameters:

Name	Type	Description	Default
relpath	str	Path relative to the container's /app directory.	required
server_name	str	An application-defined name for the MCP server. Must be a valid Python module name.	required
server_params	dict[str, Any]	MCP server configuration. stdio server configurations must specify at least a command key, sse server configurations must specify at least a url key.	required

Returns:

Type	Description
list[str]	List of tool names provided by the MCP server. Tool names are sanitized to ensure they can be used as Python module names.

Source code in ipybox/resource/client.py

async def generate_mcp_sources(self, relpath: str, server_name: str, server_params: dict[str, Any]) -> list[str]:
    """Generate Python client functions for tools provided by an MCP server.

    One MCP client function is generated per MCP tool from its metadata. The generated function is stored in a
    module named `/app/{relpath}/{server_name}/{tool_name}.py`. Importing this module and calling the function
    invokes the corresponding MCP tool. This works for both `stdio` and `sse` based MCP servers. `stdio` based
    MCP servers are executed inside the container, `sse` based MCP servers are expected to run elsewhere.

    Args:
        relpath: Path relative to the container's `/app` directory.
        server_name: An application-defined name for the MCP server. Must be a valid Python module name.
        server_params: MCP server configuration. `stdio` server configurations must specify at least a `command`
            key, `sse` server configurations must specify at least a `url` key.

    Returns:
        List of tool names provided by the MCP server. Tool names are sanitized to ensure they
            can be used as Python module names.
    """
    url = f"{self._base_url}/mcp/{relpath}/{server_name}"
    async with self._session.put(url, json=server_params) as response:
        response.raise_for_status()
        return await response.json()

get_mcp_sources async

get_mcp_sources(
    relpath: str, server_name: str
) -> dict[str, str]

Get the source code of generated MCP client functions for given MCP server_name.

Parameters:

Name	Type	Description	Default
relpath	str	Path relative to the container's /app directory	required
server_name	str	Application-defined name of an MCP server	required

Returns:

Type	Description
dict[str, str]	Source code of generated MCP client functions. Keys are tool names, values are generated sources.

Source code in ipybox/resource/client.py

async def get_mcp_sources(self, relpath: str, server_name: str) -> dict[str, str]:
    """Get the source code of generated MCP client functions for given MCP `server_name`.

    Args:
        relpath: Path relative to the container's `/app` directory
        server_name: Application-defined name of an MCP server

    Returns:
        Source code of generated MCP client functions. Keys are tool names, values are generated sources.
    """
    url = f"{self._base_url}/mcp/{relpath}"
    async with self._session.get(url, params={"server_name": server_name}) as response:
        response.raise_for_status()
        return await response.json()

get_module_sources async

get_module_sources(
    module_names: list[str],
) -> dict[str, str]

Get the source code of Python modules on the container's Python path.

Parameters:

Name	Type	Description	Default
module_names	list[str]	A list of Python module names.	required

Returns:

Type	Description
dict[str, str]	Source code of Python modules. Keys are module names, values are module sources.

Source code in ipybox/resource/client.py

async def get_module_sources(self, module_names: list[str]) -> dict[str, str]:
    """Get the source code of Python modules on the container's Python path.

    Args:
        module_names: A list of Python module names.

    Returns:
        Source code of Python modules. Keys are module names, values are module sources.
    """
    url = f"{self._base_url}/modules"
    async with self._session.get(url, params={"q": module_names}) as response:
        response.raise_for_status()
        return await response.json()

ConnectionError

Bases: Exception

Raised when a connection to a resource server cannot be established.