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: RequestState, lifespan: LifespanState, _progress_callback: Callable[[str, int, int], Awaitable[None]] | None = None, _log_callback: Callable[[str, str, Dict[str, Any]], Awaitable[None]] | None = None, _server: MCPServer | None = None)

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.

__init__(request: RequestState, lifespan: LifespanState, _progress_callback: Callable[[str, int, int], Awaitable[None]] | None = None, _log_callback: Callable[[str, str, Dict[str, Any]], Awaitable[None]] | None = None, _server: MCPServer | None = None) → None

check_cancellation() → None

Check if request has been cancelled, raise if so.

classmethod create(request_id: str, method: str, lifespan_state: LifespanState, progress_token: str | None = None, tool_invocation_id: str | None = None, metadata: Dict[str, Any] | None = None, progress_callback: Callable[[str, int, int], Awaitable[None]] | None = None, log_callback: Callable[[str, str, Dict[str, Any]], Awaitable[None]] | None = None) → Self

Create a new context for a request.

async error(message: str, **kwargs: Any) → None

Log error message.

async execute_r_with_session(script: str, args: Dict[str, Any], use_session: bool = True) → Dict[str, Any]

Execute R script with optional session support.

Parameters:

• script (str) – R script to execute

• args (Dict[str, Any]) – Arguments to pass to script

• use_session (bool) – Whether to use session (if available) or run statelessly

Returns:

Script execution results

Return type:

Dict[str, Any]

get_cache_path(key: str) → Path | None

Get cache path for key if caching is enabled.

async get_or_create_r_session(working_directory: Path | None = None) → str | None

Get or create an R session for this context.

get_r_session_id() → str | None

Get the R session ID for this context.

async info(message: str, **kwargs: Any) → None

Log info message.

is_path_allowed(path: Path) → bool

Check if path access is allowed.

is_r_session_enabled() → bool

Check if R session management is enabled.

async log(level: str, message: str, **kwargs: Any) → None

Send structured log notification.

async progress(message: str, current: int, total: int) → None

Send progress notification if progress token is available.

require_path_access(path: Path) → None

Require path access, raise if denied.

set_r_session_id(session_id: str) → None

Set the R session ID for this context.

async warn(message: str, **kwargs: Any) → None

Log warning message.

request: RequestState

lifespan: LifespanState

rmcp.create_server(name: str = 'RMCP MCP Server', version: str = None, description: str = 'RMCP provides 44 comprehensive statistical analysis tools through R:\n\n**Regression & Econometrics (8 tools):**\n- Linear/logistic regression with diagnostics and residual analysis\n- Panel data regression (fixed/random effects) with robust standard errors\n- Instrumental variables (2SLS) regression for causal inference\n- Vector autoregression (VAR) models for multivariate time series\n- Correlation analysis with significance testing and confidence intervals\n\n**Time Series Analysis (6 tools):**\n- ARIMA modeling with automatic order selection and forecasting\n- Time series decomposition (trend, seasonal, remainder components)\n- Stationarity testing (ADF, KPSS, Phillips-Perron tests)\n- Lag/lead variable creation and differencing transformations\n\n**Statistical Testing (5 tools):**\n- T-tests (one-sample, two-sample, paired) with effect sizes\n- ANOVA (one-way, two-way) with post-hoc comparisons\n- Chi-square tests for independence and goodness-of-fit\n- Normality tests (Shapiro-Wilk, Kolmogorov-Smirnov, Anderson-Darling)\n\n**Data Analysis & Transformation (9 tools):**\n- Comprehensive descriptive statistics with distribution analysis\n- Outlier detection using multiple methods (IQR, Z-score, Mahalanobis)\n- Data standardization (z-score, min-max, robust scaling)\n- Winsorization for outlier treatment and data cleaning\n- Professional frequency tables with percentages and cumulative statistics\n\n**Machine Learning (4 tools):**\n- K-means clustering with optimal cluster selection and visualization\n- Decision trees for classification and regression with pruning\n- Random forest models with variable importance and out-of-bag error\n\n**Professional Visualizations (6 tools):**\n- Scatter plots with trend lines, confidence bands, and grouping\n- Time series plots for single/multiple variables with forecasting\n- Histograms with density overlays and distribution fitting\n- Correlation heatmaps with hierarchical clustering\n- Box plots for distribution comparison and outlier identification\n- Comprehensive residual diagnostic plots (4-panel analysis)\n\n**File Operations (3 tools):**\n- CSV/Excel/JSON import with automatic type detection\n- Data filtering, export, and comprehensive dataset information\n- Missing value analysis and data quality reporting\n\n**Advanced Features:**\n- Formula builder: Convert natural language to R statistical formulas\n- Error recovery: Intelligent error diagnosis with suggested fixes\n- Flexible R execution: Custom R code with 80+ whitelisted packages\n- Example datasets: Built-in datasets for testing and learning\n\nAll tools provide professionally formatted output with markdown tables, statistical interpretations, and inline visualizations (base64 images). Results include both raw data and formatted summaries using broom/knitr for publication-ready output.') → MCPServer

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

Returns:

Configured MCPServer instance ready for configuration and startup

Return type:

MCPServer

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: Callable[[list[str] | None], None] | None = None)

Bases: object

Registry for MCP tools with schema validation.

__init__(on_list_changed: Callable[[list[str] | None], None] | None = None)

async call_tool(context: Context, name: str, arguments: dict[str, Any]) → dict[str, Any]

Call a tool with validation.

async list_tools(context: Context, cursor: str | None = None, limit: int | None = None) → dict[str, Any]

List available tools for MCP tools/list.

register(name: str, handler: Callable[[Context, dict[str, Any]], Awaitable[dict[str, Any]]], input_schema: dict[str, Any], output_schema: dict[str, Any] | None = None, title: str | None = None, description: str | None = None, annotations: dict[str, Any] | None = None) → None

Register a tool with the registry.

class rmcp.ResourcesRegistry(on_list_changed: Callable[[List[str] | None], None] | None = None)

Bases: object

Registry for MCP resources with VFS security.

__init__(on_list_changed: Callable[[List[str] | None], None] | None = None)

async list_resources(context: Context, cursor: str | None = None, limit: int | None = None) → Dict[str, Any]

List available resources for MCP resources/list.

async read_resource(context: Context, uri: str) → Dict[str, Any]

Read a resource for MCP resources/read.

register_memory_object(name: str, data: Any, description: str | None = None, mime_type: str = 'application/json') → None

Register an in-memory object as a resource.

register_resource_template(uri_template: str, name: str, description: str | None = None) → None

Register a parameterized resource template.

register_static_resource(uri: str, name: str, description: str | None = None, mime_type: str | None = None, content_loader: str | bytes | Callable[[], Any] | Callable[[], Awaitable[Any]] | None = None) → None

Register a static resource.

class rmcp.PromptsRegistry(on_list_changed: Callable[[List[str] | None], None] | None = None)

Bases: object

Registry for MCP prompts with templating support.

__init__(on_list_changed: Callable[[List[str] | None], None] | None = None)

async get_prompt(context: Context, name: str, arguments: Dict[str, Any] | None = None) → Dict[str, Any]

Get a rendered prompt for MCP prompts/get.

async list_prompts(context: Context, cursor: str | None = None, limit: int | None = None) → Dict[str, Any]

List available prompts for MCP prompts/list.

register(name: str, title: str, description: str, template: str, arguments_schema: Dict[str, Any] | None = None, annotations: Dict[str, Any] | None = None) → None

Register a prompt template.

rmcp.tool(name: str, input_schema: dict[str, Any], output_schema: dict[str, Any] | None = None, title: str | None = None, description: str | None = None, annotations: dict[str, Any] | None = None)

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: str, name: str, description: str | None = None, mime_type: str | None = None)

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: str, title: str, description: str, arguments_schema: Dict[str, Any] | None = None, annotations: Dict[str, Any] | None = None)

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

rmcp.r_integration.get_r_binary_path() → str

Discover and cache the R binary path.

Returns:

Path to R binary

Return type:

str

Raises:

FileNotFoundError – If R binary cannot be found

exception rmcp.r_integration.RExecutionError(message: str, stdout: str = '', stderr: str = '', returncode: int = None)

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: str, stdout: str = '', stderr: str = '', returncode: int = None)

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

rmcp.r_integration.check_r_version() → tuple[bool, str]

Check if R version is 4.4.0 or higher.

Returns:

Tuple of (is_compatible, version_string) - is_compatible: True if R version >= 4.4.0 - version_string: Full R version string for logging

Raises:

RExecutionError – If R is not available or version check fails

Return type:

tuple[bool, str]

rmcp.r_integration.execute_r_script(script: str, args: dict[str, Any]) → dict[str, Any]

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 :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.

Returns:

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

Raises:

• RExecutionError – If R script execution fails, with detailed error info

• FileNotFoundError – If R is not installed or not in PATH

• json.JSONDecodeError – If R script produces invalid JSON output

Return type:

dict[str, Any]

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: str, args: dict[str, Any], context=None) → dict[str, Any]

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 :param script: R script code to execute :param args: Arguments to pass to the R script as JSON :param context: Optional context for progress reporting and logging

Returns:

Result data from R script execution

Return type:

dict[str, Any]

Raises:

• RExecutionError – If R script execution fails

• asyncio.CancelledError – If the operation is cancelled

rmcp.r_integration.get_r_image_encoder_script() → str

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: str, args: dict[str, Any], include_image: bool = True, image_width: int = 800, image_height: int = 600) → dict[str, Any]

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. :param script: R script code to execute :param args: Arguments to pass to R script :param include_image: Whether to attempt image capture and encoding :param image_width: Width of captured image in pixels :param image_height: Height of captured image in pixels

Returns:

Dict containing R script results, optionally with image_data and image_mime_type

Return type:

dict[str, Any]

async rmcp.r_integration.execute_r_script_with_image_async(script: str, args: dict[str, Any], include_image: bool = True, image_width: int = 800, image_height: int = 600) → dict[str, Any]

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. :param script: R script code to execute :param args: Arguments to pass to R script :param include_image: Whether to attempt image capture and encoding :param image_width: Width of captured image in pixels :param image_height: Height of captured image in pixels

Returns:

Dict containing R script results, optionally with image_data and image_mime_type

Return type:

dict[str, Any]

rmcp.r_integration.diagnose_r_installation() → dict[str, Any]

Diagnose R installation and return comprehensive status information.

Returns:

• r_available: Whether R binary is found

• r_path: Path to R binary

• r_version: R version string (if available)

• jsonlite_available: Whether jsonlite package is installed

• environment: Relevant environment variables

• error: Any error encountered during diagnosis

Return type:

dict containing diagnostic information including

async rmcp.r_integration.diagnose_r_installation_async() → dict[str, Any]

Asynchronous version of R installation diagnosis.

Returns:

dict containing the same diagnostic information as diagnose_r_installation

Return type:

dict[str, Any]