rmcp package

RMCP MCP Server - A Model Context Protocol server for R-based statistical analysis. This package implements a production-ready MCP server following established patterns: - Spec correctness by construction using official SDK - Clean separation of concerns (protocol/registries/domain) - Security by default (VFS, allowlists, sandboxing) - Transport-agnostic design (stdio primary, HTTP optional) - Explicit schemas and typed context objects

class rmcp.Context(request, lifespan, _progress_callback=None, _log_callback=None)[source]

Bases: object

Typed context passed to all tool handlers. Provides both per-request state and shared lifespan state, plus helpers for logging, progress, and cancellation.

request: RequestState
lifespan: LifespanState
classmethod create(request_id, method, lifespan_state, progress_token=None, tool_invocation_id=None, metadata=None, progress_callback=None, log_callback=None)[source]

Create a new context for a request.

Return type:

Context

async progress(message, current, total)[source]

Send progress notification if progress token is available.

Return type:

None

async log(level, message, **kwargs)[source]

Send structured log notification.

Return type:

None

async info(message, **kwargs)[source]

Log info message.

Return type:

None

async warn(message, **kwargs)[source]

Log warning message.

Return type:

None

async error(message, **kwargs)[source]

Log error message.

Return type:

None

check_cancellation()[source]

Check if request has been cancelled, raise if so.

Return type:

None

is_path_allowed(path)[source]

Check if path access is allowed.

Return type:

bool

require_path_access(path)[source]

Require path access, raise if denied.

Return type:

None

get_cache_path(key)[source]

Get cache path for key if caching is enabled.

Return type:

Optional[Path]

__init__(request, lifespan, _progress_callback=None, _log_callback=None)
rmcp.create_server(name='RMCP MCP Server', version=None, description='R-based statistical analysis MCP server')[source]

Factory function to create a new MCP server instance. :type name: str :param name: Human-readable server name :type version: str :param version: Semantic version string :type description: str :param description: Brief description of server capabilities

Return type:

MCPServer

Returns:

Configured MCPServer instance ready for configuration and startup

Example

>>> server = create_server(
...     name="My Analytics Server",
...     version="1.0.0",
...     description="Custom R analytics tools"
... )
>>> server.configure(allowed_paths=["/data"])
class rmcp.ToolsRegistry(on_list_changed=None)[source]

Bases: object

Registry for MCP tools with schema validation.

__init__(on_list_changed=None)[source]
register(name, handler, input_schema, output_schema=None, title=None, description=None, annotations=None)[source]

Register a tool with the registry.

Return type:

None

async list_tools(context, cursor=None, limit=None)[source]

List available tools for MCP tools/list.

Return type:

dict[str, Any]

async call_tool(context, name, arguments)[source]

Call a tool with validation.

Return type:

dict[str, Any]

class rmcp.ResourcesRegistry(on_list_changed=None)[source]

Bases: object

Registry for MCP resources with VFS security.

__init__(on_list_changed=None)[source]
register_static_resource(uri, name, description=None, mime_type=None, content_loader=None)[source]

Register a static resource.

Return type:

None

register_memory_object(name, data, description=None, mime_type='application/json')[source]

Register an in-memory object as a resource.

Return type:

None

register_resource_template(uri_template, name, description=None)[source]

Register a parameterized resource template.

Return type:

None

async list_resources(context, cursor=None, limit=None)[source]

List available resources for MCP resources/list.

Return type:

Dict[str, Any]

async read_resource(context, uri)[source]

Read a resource for MCP resources/read.

Return type:

Dict[str, Any]

class rmcp.PromptsRegistry(on_list_changed=None)[source]

Bases: object

Registry for MCP prompts with templating support.

__init__(on_list_changed=None)[source]
register(name, title, description, template, arguments_schema=None, annotations=None)[source]

Register a prompt template.

Return type:

None

async list_prompts(context, cursor=None, limit=None)[source]

List available prompts for MCP prompts/list.

Return type:

Dict[str, Any]

async get_prompt(context, name, arguments=None)[source]

Get a rendered prompt for MCP prompts/get.

Return type:

Dict[str, Any]

rmcp.tool(name, input_schema, output_schema=None, title=None, description=None, annotations=None)[source]

Decorator to register a function as an MCP tool. Usage:

@tool(

name=”analyze_data”, input_schema={

“type”: “object”, “properties”: {

“data”: table_schema(), “method”: choice_schema([“mean”, “median”, “mode”])

}, “required”: [“data”]

}, description=”Analyze dataset with specified method”

) async def analyze_data(context: Context, params: dict[str, Any]) -> dict[str, Any]:

# Tool implementation return {“result”: “analysis complete”}

rmcp.resource(uri, name, description=None, mime_type=None)[source]

Decorator to register a static resource. Usage:

@resource(

uri=”static://example”, name=”Example Resource”, description=”An example static resource”

) def example_resource():

return “resource content”

rmcp.prompt(name, title, description, arguments_schema=None, annotations=None)[source]

Decorator to register a prompt template. Usage:

@prompt(

name=”analyze_workflow”, title=”Statistical Analysis Workflow”, description=”Guide for comprehensive statistical analysis”, arguments_schema={

“type”: “object”, “properties”: {

“dataset_name”: {“type”: “string”}, “analysis_type”: {“type”: “string”, “enum”: [“descriptive”, “inferential”, “predictive”]}

}, “required”: [“dataset_name”]

}

) def analyze_workflow():

return ‘’’ I’ll help you analyze the {dataset_name} dataset using {analysis_type} methods. Let me start by examining the data structure and then proceed with the analysis. ‘’’

Subpackages

Submodules

rmcp.cli module

Command-line interface for RMCP MCP Server. Provides entry points for running the server with different transports and configurations, following the principle of “multiple deployment targets.”

rmcp.r_integration module

R Integration Module for RMCP Statistical Analysis. This module provides a clean interface for executing R scripts from Python, handling data serialization, error management, and resource cleanup. Key features: - JSON-based data exchange between Python and R - Automatic temporary file management - Comprehensive error handling with detailed diagnostics - Timeout protection for long-running R operations - Cross-platform R execution support .. rubric:: Example

>>> script = '''
... result <- list(
...     mean_value = mean(args$data),
...     std_dev = sd(args$data)
... )
... '''
>>> args = {"data": [1, 2, 3, 4, 5]}
>>> result = execute_r_script(script, args)
>>> print(result["mean_value"])  # 3.0
exception rmcp.r_integration.RExecutionError(message, stdout='', stderr='', returncode=None)[source]

Bases: Exception

Exception raised when R script execution fails. This exception provides detailed information about R execution failures, including stdout/stderr output and process return codes for debugging. .. attribute:: message

Human-readable error description

stdout

Standard output from R process (if any)

stderr

Standard error from R process (if any)

returncode

Process exit code (if available)

Example

>>> try:
...     execute_r_script("invalid R code", {})
... except RExecutionError as e:
...     print(f"R failed: {e}")
...     print(f"Error details: {e.stderr}")
__init__(message, stdout='', stderr='', returncode=None)[source]

Initialize R execution error. :type message: str :param message: Primary error message :type stdout: str :param stdout: R process standard output :type stderr: str :param stderr: R process standard error :type returncode: int :param returncode: R process exit code

rmcp.r_integration.execute_r_script(script, args)[source]

Execute an R script with arguments and return JSON results. This function creates a complete R execution environment by: 1. Writing arguments to a temporary JSON file 2. Creating an R script that loads jsonlite and reads the arguments 3. Appending the user’s R code 4. Writing results to a JSON output file 5. Executing R and parsing the results 6. Cleaning up all temporary files :type script: str :param script: R code to execute. Must set a ‘result’ variable with output.

The script has access to an ‘args’ variable containing the arguments.

Parameters:

args (dict[str, Any]) – Dictionary of arguments available to R script as ‘args’ variable. All values must be JSON-serializable.

Return type:

dict[str, Any]

Returns:

Dictionary containing the R script results (contents of ‘result’ variable).

Raises:

Example

>>> # Calculate statistics on a dataset
>>> r_code = '''
... result <- list(
...     mean = mean(args$values),
...     median = median(args$values),
...     sd = sd(args$values)
... )
... '''
>>> args = {"values": [1, 2, 3, 4, 5]}
>>> stats = execute_r_script(r_code, args)
>>> print(stats["mean"])  # 3.0
>>> # Linear regression example
>>> r_code = '''
... df <- data.frame(args$data)
... model <- lm(y ~ x, data = df)
... result <- list(
...     coefficients = coef(model),
...     r_squared = summary(model)$r.squared
... )
... '''
>>> data = {"data": {"x": [1,2,3,4], "y": [2,4,6,8]}}
>>> reg_result = execute_r_script(r_code, data)
async rmcp.r_integration.execute_r_script_async(script, args, context=None)[source]

Execute R script asynchronously with proper cancellation support and concurrency control. This function provides: - True async execution using asyncio.create_subprocess_exec - Proper subprocess cancellation (SIGTERM -> SIGKILL) - Global concurrency limiting via semaphore - Progress reporting from R scripts via context - Same interface and error handling as execute_r_script :type script: str :param script: R script code to execute :type args: dict[str, Any] :param args: Arguments to pass to the R script as JSON :type context: :param context: Optional context for progress reporting and logging

Returns:

Result data from R script execution

Return type:

dict[str, Any]

Raises:
rmcp.r_integration.get_r_image_encoder_script()[source]

Get R script code for encoding plots as base64 images. This function returns R code that can be included in visualization scripts to generate base64-encoded PNG images for display in Claude. :returns: R script code with base64 encoding functions :rtype: str

rmcp.r_integration.execute_r_script_with_image(script, args, include_image=True, image_width=800, image_height=600)[source]

Execute R script and optionally include base64-encoded image data. This function extends execute_r_script to support automatic image encoding for visualization tools. If include_image is True, it will attempt to capture any plot generated by the R script and return it as base64-encoded PNG data. :type script: str :param script: R script code to execute :type args: dict[str, Any] :param args: Arguments to pass to R script :type include_image: bool :param include_image: Whether to attempt image capture and encoding :type image_width: int :param image_width: Width of captured image in pixels :type image_height: int :param image_height: Height of captured image in pixels

Return type:

dict[str, Any]

Returns:

Dict containing R script results, optionally with image_data and image_mime_type

async rmcp.r_integration.execute_r_script_with_image_async(script, args, include_image=True, image_width=800, image_height=600)[source]

Execute R script asynchronously and optionally include base64-encoded image data. This function extends execute_r_script_async to support automatic image encoding for visualization tools. If include_image is True, it will attempt to capture any plot generated by the R script and return it as base64-encoded PNG data. :type script: str :param script: R script code to execute :type args: dict[str, Any] :param args: Arguments to pass to R script :type include_image: bool :param include_image: Whether to attempt image capture and encoding :type image_width: int :param image_width: Width of captured image in pixels :type image_height: int :param image_height: Height of captured image in pixels

Return type:

dict[str, Any]

Returns:

Dict containing R script results, optionally with image_data and image_mime_type