API Reference

Complete API documentation for RMCP components.

Core Modules

Server

MCP Server shell with lifecycle hooks. This module provides the main server class that: - Initializes the MCP app using official SDK - Manages lifespan hooks (startup/shutdown) - Composes transports at the edge - Centralizes registry management Following the principle: “A single shell centralizes initialization and teardown.”

class rmcp.core.server.MCPServer(name: str = 'RMCP MCP Server', version: str | None = 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.')

Bases: object

Main MCP server shell that manages lifecycle and registries. This class serves as the central orchestrator for the RMCP MCP server, providing: - Lifespan management (startup/shutdown hooks) - Registry composition (tools/resources/prompts) - Security policy enforcement via VFS - Transport-agnostic request handling - Request tracking and cancellation support The server follows the Model Context Protocol (MCP) specification for communication with AI assistants like Claude Desktop. .. rubric:: Example

>>> server = MCPServer(name="My Server", version="1.0.0")
>>> server.configure(allowed_paths=["/data"], read_only=True)
>>> # Register tools, prompts, resources...
>>> await server.startup()

__init__(name: str = 'RMCP MCP Server', version: str | None = 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.')

Initialize the MCP server instance. :param name: Human-readable name for the server :param version: Semantic version string :param description: Brief description of server capabilities

configure(allowed_paths: list[str] | None = None, cache_root: str | None = None, read_only: bool = True, **settings: Any) → MCPServer

Configure server security and operational settings.

Parameters:

• allowed_paths (list[str] | None) – List of filesystem paths the server can access. If None, defaults to current working directory.

• cache_root (str | None) – Directory for caching intermediate results. Created if it doesn’t exist.

• read_only (bool) – Whether filesystem access is read-only. Recommended for production deployments.

• **settings (Any) – Additional configuration options passed to lifespan state.

Returns:

Self for method chaining.

Return type:

MCPServer

Example

>>> server.configure(
...     allowed_paths=["/data", "/models"],
...     cache_root="/tmp/rmcp_cache",
...     read_only=True
... )

on_startup(func: Callable[[], Awaitable[None]]) → Callable[[], Awaitable[None]]

Register a callback to run during server startup. :param func: Async function to call during startup. Should not take arguments.

Returns:

The same function (for use as decorator).

Return type:

Callable[[], Awaitable[None]]

Example

>>> @server.on_startup
... async def initialize_r_packages():
...     # Check R installation, load packages, etc.
...     pass

on_shutdown(func: Callable[[], Awaitable[None]]) → Callable[[], Awaitable[None]]

Register a callback to run during server shutdown. :param func: Async function to call during shutdown. Should not take arguments.

Returns:

The same function (for use as decorator).

Return type:

Callable[[], Awaitable[None]]

Example

>>> @server.on_shutdown
... async def cleanup_temp_files():
...     # Clean up R temporary files, connections, etc.
...     pass

async startup() → None

Start the server and run all startup callbacks. This method should be called once before handling any requests. It executes all registered startup callbacks in registration order. :raises Exception: If any startup callback fails, the exception propagates.

async shutdown() → None

Shutdown the server gracefully. This method: 1. Cancels all active requests 2. Runs all shutdown callbacks (continuing on errors) 3. Logs completion Shutdown callbacks are called in registration order and errors are logged but don’t prevent other callbacks from running.

create_context(request_id: str, method: str, progress_token: str | None = None, tool_invocation_id: str | None = None, metadata: dict[str, Any] | None = None) → Context

Create execution context for a request. :param request_id: Unique identifier for the request :param method: MCP method being called (e.g., “tools/call”) :param progress_token: Optional token for progress reporting

Returns:

Context object with progress/logging callbacks configured

Return type:

Context

finish_request(request_id: str) → None

Clean up request tracking after completion. :param request_id: The request ID to remove from active tracking

async cancel_request(request_id: str) → None

Cancel an active request by ID. :param request_id: The request ID to cancel

Note

If the request is not found, this method does nothing. Cancellation is cooperative - the request handler must check for cancellation periodically.

async handle_request(request: dict[str, Any]) → dict[str, Any] | None

Handle incoming MCP request and route to appropriate handler. This is the main entry point for all MCP requests. It: 1. Extracts method, ID, and parameters from the request 2. Routes to appropriate registry (tools, resources, prompts) 3. Returns properly formatted JSON-RPC response 4. Handles errors with appropriate error codes :param request: JSON-RPC request dict with method, id, and params

Returns:

JSON-RPC response dict or None for notifications

Return type:

dict[str, Any] | None

Supported methods:

• initialize: Initialize MCP connection and return capabilities

• tools/list: List available tools

• tools/call: Execute a tool with parameters

• resources/list: List available resources

• resources/read: Read a resource by URI

• prompts/list: List available prompts

• prompts/get: Get a prompt with arguments

• completion/complete: Provide auto-completion suggestions

• logging/setLevel: Set the server logging level

create_message_handler(transport: Transport) → Callable[[dict[str, Any]], Awaitable[dict[str, Any] | None]]

Bind a transport to the server handler so context can emit feedback.

rmcp.core.server.create_server(name: str = 'RMCP MCP Server', version: str | None = 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"])

Context Management

Typed context object for MCP requests. The Context object provides: - Per-request state (request ID, progress token, cancellation) - Lifespan state (settings, caches, resources) - Cross-cutting features (logging, progress, security) Following the principle: “Makes cross-cutting features universal without globals.”

class rmcp.core.context.RequestState(request_id: str, method: str, progress_token: str | None = None, tool_invocation_id: str | None = None, metadata: dict[str, ~typing.Any] = <factory>, cancelled: bool = False)

Bases: object

Per-request state passed to tool handlers.

request_id: str

method: str

progress_token: str | None = None

tool_invocation_id: str | None = None

metadata: dict[str, Any]

cancelled: bool = False

is_cancelled() → bool

Check if request has been cancelled.

cancel() → None

Mark request as cancelled.

__init__(request_id: str, method: str, progress_token: str | None = None, tool_invocation_id: str | None = None, metadata: dict[str, ~typing.Any] = <factory>, cancelled: bool = False) → None

class rmcp.core.context.LifespanState(settings: dict[str, ~typing.Any] = <factory>, allowed_paths: list[~pathlib.Path] = <factory>, read_only: bool = True, cache_root: ~pathlib.Path | None = None, content_cache: dict[str, ~typing.Any] = <factory>, resource_mounts: dict[str, ~pathlib.Path] = <factory>, vfs: ~typing.Any | None = None, current_log_level: str = 'info', r_session_enabled: bool = False, r_session_timeout: float = 3600.0, default_r_session_id: str | None = None)

Bases: object

Lifespan state shared across requests.

settings: dict[str, Any]

allowed_paths: list[Path]

read_only: bool = True

cache_root: Path | None = None

content_cache: dict[str, Any]

resource_mounts: dict[str, Path]

vfs: Any | None = None

current_log_level: str = 'info'

r_session_enabled: bool = False

r_session_timeout: float = 3600.0

default_r_session_id: str | None = None

__init__(settings: dict[str, ~typing.Any] = <factory>, allowed_paths: list[~pathlib.Path] = <factory>, read_only: bool = True, cache_root: ~pathlib.Path | None = None, content_cache: dict[str, ~typing.Any] = <factory>, resource_mounts: dict[str, ~pathlib.Path] = <factory>, vfs: ~typing.Any | None = None, current_log_level: str = 'info', r_session_enabled: bool = False, r_session_timeout: float = 3600.0, default_r_session_id: str | None = None) → None

class rmcp.core.context.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.

request: RequestState

lifespan: LifespanState

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 progress(message: str, current: int, total: int) → None

Send progress notification if progress token is available.

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

Send structured log notification.

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

Log info message.

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

Log warning message.

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

Log error message.

check_cancellation() → None

Check if request has been cancelled, raise if so.

is_path_allowed(path: Path) → bool

Check if path access is allowed.

require_path_access(path: Path) → None

Require path access, raise if denied.

get_cache_path(key: str) → Path | None

Get cache path for key if caching is enabled.

is_r_session_enabled() → bool

Check if R session management is enabled.

get_r_session_id() → str | None

Get the R session ID for this context.

set_r_session_id(session_id: str) → None

Set the R session ID for this context.

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

Get or create an R session for this context.

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]

__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

Statistical Tools

Regression Analysis

Regression Analysis Tools for RMCP MCP Server. This module provides comprehensive regression modeling capabilities including: - Linear regression with diagnostics - Logistic regression for binary outcomes - Correlation analysis with significance testing - Comprehensive model validation and statistics All tools support missing value handling, weighted observations, and return detailed statistical outputs suitable for research and business analysis. Example Usage:

>>> # Linear regression on sales data
>>> data = {"sales": [100, 120, 140], "advertising": [10, 15, 20]}
>>> result = await linear_model(context, {
...     "data": data,
...     "formula": "sales ~ advertising"
... })
>>> print(f"R-squared: {result['r_squared']}")

async rmcp.tools.regression.linear_model(context, params) → dict[str, Any]

Fit ordinary least squares (OLS) linear regression model.

This tool performs comprehensive linear regression analysis using R’s lm() function. It supports weighted regression, missing value handling, and returns detailed model diagnostics including coefficients, significance tests, and goodness-of-fit.

Parameters:

• context – Request execution context for logging and progress

• params –

  Dictionary containing:

  • data: Dataset as dict of column_name -> [values]

  • formula: R formula string (e.g., “y ~ x1 + x2”)

  • weights: Optional array of observation weights

  • na_action: How to handle missing values (“na.omit”, “na.exclude”, “na.fail”)

Returns:

• coefficients: Model coefficients by variable name

• std_errors: Standard errors of coefficients

• t_values: t-statistics for coefficient tests

• p_values: p-values for coefficient significance

• r_squared: Coefficient of determination

• adj_r_squared: Adjusted R-squared

• fstatistic: Overall F-statistic value

• f_pvalue: p-value for overall model significance

• residual_se: Residual standard error

• fitted_values: Predicted values for each observation

• residuals: Model residuals

• n_obs: Number of observations used

Return type:

Dictionary containing

Example

>>> # Simple linear regression
>>> data = {
...     "price": [100, 120, 140, 160, 180],
...     "size": [1000, 1200, 1400, 1600, 1800]
... }
>>> result = await linear_model(context, {
...     "data": data,
...     "formula": "price ~ size"
... })
>>> print(f"Price increases ${result['coefficients']['size']:.2f} per sq ft")
>>> print(f"Model explains {result['r_squared']:.1%} of variance")
>>> # Multiple regression with weights
>>> data = {
...     "sales": [100, 150, 200, 250],
...     "advertising": [10, 20, 30, 40],
...     "price": [50, 45, 40, 35]
... }
>>> result = await linear_model(context, {
...     "data": data,
...     "formula": "sales ~ advertising + price",
...     "weights": [1, 1, 2, 2]  # Weight later observations more
... })

async rmcp.tools.regression.correlation_analysis(context, params) → dict[str, Any]

Compute correlation matrix with significance testing.

This tool calculates pairwise correlations between numeric variables using Pearson, Spearman, or Kendall methods. It includes significance tests for each correlation and handles missing values appropriately.

Parameters:

• context – Request execution context for logging and progress

• params –

  Dictionary containing:

  • data: Dataset as dict of column_name -> [values]

  • variables: Optional list of variable names to include

  • method: Correlation method (“pearson”, “spearman”, “kendall”)

  • use: Missing value handling strategy

Returns:

• correlation_matrix: Pairwise correlations as nested dict

• significance_tests: p-values for each correlation

• sample_sizes: Number of complete observations for each pair

• variables_used: List of variables included in analysis

• method_used: Correlation method applied

Return type:

Dictionary containing

Example

>>> # Basic correlation analysis
>>> data = {
...     "sales": [100, 150, 200, 250, 300],
...     "advertising": [10, 20, 25, 35, 40],
...     "price": [50, 48, 45, 42, 40]
... }
>>> result = await correlation_analysis(context, {
...     "data": data,
...     "method": "pearson"
... })
>>> sales_ad_corr = result["correlation_matrix"]["sales"]["advertising"]
>>> print(f"Sales-Advertising correlation: {sales_ad_corr:.3f}")
>>> # Spearman correlation for non-linear relationships
>>> result = await correlation_analysis(context, {
...     "data": data,
...     "method": "spearman",
...     "variables": ["sales", "advertising"]
... })

async rmcp.tools.regression.logistic_regression(context, params) → dict[str, Any]

Fit logistic regression model.

Visualization Tools

Visualization tools for RMCP. Statistical plotting and data visualization capabilities.

async rmcp.tools.visualization.scatter_plot(context, params) → dict[str, Any]

Create scatter plot.

async rmcp.tools.visualization.histogram(context, params) → dict[str, Any]

Create histogram.

async rmcp.tools.visualization.boxplot(context, params) → dict[str, Any]

Create box plot.

async rmcp.tools.visualization.time_series_plot(context, params) → dict[str, Any]

Create time series plot.

async rmcp.tools.visualization.correlation_heatmap(context, params) → dict[str, Any]

Create correlation heatmap.

async rmcp.tools.visualization.regression_plot(context, params) → dict[str, Any]

Create regression diagnostic plots.

Time Series Analysis

Time series analysis tools for RMCP. Comprehensive time series modeling and forecasting capabilities.

async rmcp.tools.timeseries.arima_model(context, params) → dict[str, Any]

Fit ARIMA model and generate forecasts.

async rmcp.tools.timeseries.decompose_timeseries(context, params) → dict[str, Any]

Decompose time series into components.

async rmcp.tools.timeseries.stationarity_test(context, params) → dict[str, Any]

Test time series stationarity.

Machine Learning

Machine learning tools for RMCP. Clustering, classification trees, and ML capabilities.

async rmcp.tools.machine_learning.kmeans_clustering(context, params) → dict[str, Any]

Perform K-means clustering.

async rmcp.tools.machine_learning.decision_tree(context, params) → dict[str, Any]

Build decision tree model.

async rmcp.tools.machine_learning.random_forest(context, params) → dict[str, Any]

Build Random Forest model.

Statistical Tests

Statistical hypothesis testing tools for RMCP. Comprehensive statistical testing capabilities.

async rmcp.tools.statistical_tests.t_test(context, params) → dict[str, Any]

Perform t-test analysis.

async rmcp.tools.statistical_tests.anova(context, params) → dict[str, Any]

Perform ANOVA analysis.

async rmcp.tools.statistical_tests.chi_square_test(context, params) → dict[str, Any]

Perform chi-square tests.

async rmcp.tools.statistical_tests.normality_test(context, params) → dict[str, Any]

Test for normality.

Econometrics

Econometric analysis tools for RMCP. Advanced econometric modeling for panel data, instrumental variables, etc.

async rmcp.tools.econometrics.panel_regression(context, params) → dict[str, Any]

Perform panel data regression.

async rmcp.tools.econometrics.instrumental_variables(context, params) → dict[str, Any]

Perform instrumental variables regression.

async rmcp.tools.econometrics.var_model(context, params) → dict[str, Any]

Fit Vector Autoregression model.

R Integration

R Integration Core

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 = 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 = 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

Parameters:

• script (str) – R code to execute. Must set a ‘result’ variable with output. The script has access to an ‘args’ variable containing the arguments.

• 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]

Configuration System

Configuration Models

Configuration data models for RMCP.

This module defines the complete configuration structure for RMCP with type hints, defaults, and validation. Configuration values are resolved in hierarchical order:

1. Command-line arguments (highest priority)

2. Environment variables (RMCP_* prefix)

3. User configuration file (~/.rmcp/config.json)

4. System configuration file (/etc/rmcp/config.json)

5. Built-in defaults (lowest priority)

Examples

Environment variable configuration:

export RMCP_HTTP_HOST=0.0.0.0
export RMCP_HTTP_PORT=9000
export RMCP_R_TIMEOUT=180

Configuration file (~/.rmcp/config.json):

{
  "http": {"host": "0.0.0.0", "port": 9000},
  "r": {"timeout": 180, "max_sessions": 20},
  "logging": {"level": "DEBUG"}
}

Command-line override:

rmcp --config custom.json --debug start

class rmcp.config.models.HTTPConfig(host: str = 'localhost', port: int = 8000, ssl_keyfile: str | None = None, ssl_certfile: str | None = None, ssl_keyfile_password: str | None = None, cors_origins: list[str | ~typing.Literal['*']] = <factory>)

Bases: object

HTTP transport configuration.

Controls the HTTP server behavior for RMCP when running in HTTP mode.

Environment Variables:

• RMCP_HTTP_HOST - Server binding address

• RMCP_HTTP_PORT - Server port number

• RMCP_HTTP_CORS_ORIGINS - Allowed CORS origins (comma-separated)

• RMCP_HTTP_SSL_KEYFILE - SSL private key file path

• RMCP_HTTP_SSL_CERTFILE - SSL certificate file path

Security Considerations:

• Default binding to localhost for security

• Set host="0.0.0.0" for remote access (implement authentication)

• Configure cors_origins appropriately for web clients

• SSL files must both be provided if either is specified

Examples

Development configuration:

http = HTTPConfig(
    host="localhost",
    port=8000
)

Production with SSL:

http = HTTPConfig(
    host="0.0.0.0",
    port=443,
    ssl_keyfile="/etc/ssl/private/rmcp.key",
    ssl_certfile="/etc/ssl/certs/rmcp.crt",
    cors_origins=["https://myapp.example.com"]
)

host: str

localhost for security. Use 0.0.0.0 for remote access.

Type:

Server binding address. Default

port: int

Server port number. Must be between 1-65535.

ssl_keyfile: str | None

SSL private key file path. Required if ssl_certfile is specified.

ssl_certfile: str | None

SSL certificate file path. Required if ssl_keyfile is specified.

ssl_keyfile_password: str | None

SSL private key password if the key file is encrypted.

cors_origins: list[str | Literal['*']]

Allowed CORS origins for cross-origin requests. Supports wildcards.

__init__(host: str = 'localhost', port: int = 8000, ssl_keyfile: str | None = None, ssl_certfile: str | None = None, ssl_keyfile_password: str | None = None, cors_origins: list[str | ~typing.Literal['*']] = <factory>) → None

class rmcp.config.models.RConfig(timeout: int = 120, session_timeout: int = 3600, max_sessions: int = 10, binary_path: str | None = None, version_check_timeout: int = 30)

Bases: object

R process configuration.

Controls R process execution, session management, and resource limits.

Environment Variables:

• RMCP_R_TIMEOUT - R script execution timeout (seconds)

• RMCP_R_SESSION_TIMEOUT - R session lifetime (seconds)

• RMCP_R_MAX_SESSIONS - Maximum concurrent R sessions

• RMCP_R_BINARY_PATH - Custom R binary path

• RMCP_R_VERSION_CHECK_TIMEOUT - R version check timeout

Resource Considerations:

• Limit max_sessions based on available memory (R sessions consume ~100-200MB each)

• Set reasonable timeout to prevent runaway processes

• Use session_timeout to free resources from idle sessions

Examples

Development configuration:

r = RConfig(
    timeout=300,
    max_sessions=5
)

Production configuration:

r = RConfig(
    timeout=120,
    session_timeout=1800,
    max_sessions=50,
    binary_path="/usr/local/bin/R"
)

timeout: int

R script execution timeout in seconds. Prevents runaway processes.

session_timeout: int

R session lifetime in seconds. Sessions are cleaned up after this time.

max_sessions: int

Maximum concurrent R sessions. Limit based on available memory.

binary_path: str | None

Custom R binary path. Auto-detected if None.

version_check_timeout: int

R version check timeout in seconds during startup.

__init__(timeout: int = 120, session_timeout: int = 3600, max_sessions: int = 10, binary_path: str | None = None, version_check_timeout: int = 30) → None

class rmcp.config.models.SecurityConfig(vfs_max_file_size: int = 52428800, vfs_allowed_paths: list[str] = <factory>, vfs_read_only: bool = True, vfs_allowed_mime_types: list[str] = <factory>)

Bases: object

Security and filesystem configuration.

Controls Virtual File System (VFS) security boundaries and access controls.

Environment Variables:

• RMCP_VFS_MAX_FILE_SIZE - Maximum file size in bytes

• RMCP_VFS_ALLOWED_PATHS - Additional allowed paths (comma-separated)

• RMCP_VFS_READ_ONLY - Enable read-only mode (true/false)

Security Considerations:

• Keep vfs_read_only=True in production to prevent file modification

• Restrict vfs_allowed_paths to necessary directories only

• Set appropriate vfs_max_file_size limits to prevent resource exhaustion

• Only trusted file types are allowed by default

Examples

Production security (read-only):

security = SecurityConfig(
    vfs_max_file_size=104857600,  # 100MB
    vfs_read_only=True,
    vfs_allowed_paths=["/data/readonly"]
)

Development (write access):

security = SecurityConfig(
    vfs_read_only=False,
    vfs_allowed_paths=["/tmp", "/home/user/datasets"]
)

vfs_max_file_size: int

50MB.

Type:

Maximum file size for VFS operations in bytes. Default

vfs_allowed_paths: list[str]

Additional filesystem paths accessible via VFS. Empty = temp directory only.

vfs_read_only: bool

Enable VFS read-only mode. Prevents file modification in production.

vfs_allowed_mime_types: list[str]

Allowed MIME types for file operations. Only trusted formats by default.

__init__(vfs_max_file_size: int = 52428800, vfs_allowed_paths: list[str] = <factory>, vfs_read_only: bool = True, vfs_allowed_mime_types: list[str] = <factory>) → None

class rmcp.config.models.PerformanceConfig(threadpool_max_workers: int = 2, callback_timeout: int = 300, process_cleanup_timeout: int = 5)

Bases: object

Performance and resource configuration.

Controls concurrency, timeouts, and resource management.

Environment Variables:

• RMCP_THREADPOOL_MAX_WORKERS - Max workers for stdio transport

• RMCP_CALLBACK_TIMEOUT - Bidirectional callback timeout (seconds)

• RMCP_PROCESS_CLEANUP_TIMEOUT - Process cleanup timeout (seconds)

Performance Tuning:

• Adjust threadpool_max_workers based on CPU cores

• Increase callback_timeout for slow operations

• Keep process_cleanup_timeout low for faster resource cleanup

Examples

High-performance server:

performance = PerformanceConfig(
    threadpool_max_workers=8,
    callback_timeout=600
)

Resource-constrained environment:

performance = PerformanceConfig(
    threadpool_max_workers=1,
    callback_timeout=120
)

threadpool_max_workers: int

Maximum workers for stdio transport. Adjust based on CPU cores.

__init__(threadpool_max_workers: int = 2, callback_timeout: int = 300, process_cleanup_timeout: int = 5) → None

callback_timeout: int

Bidirectional callback timeout in seconds. For slow operations.

process_cleanup_timeout: int

Process cleanup timeout in seconds. Keep low for faster cleanup.

class rmcp.config.models.LoggingConfig(level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO', format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s', stderr_output: bool = True)

Bases: object

Logging configuration.

Controls log output format and verbosity.

Environment Variables:

• RMCP_LOG_LEVEL - Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• RMCP_LOG_FORMAT - Log message format string

Log Levels:

• DEBUG - Detailed debugging information

• INFO - General operational messages

• WARNING - Warning messages about potential issues

• ERROR - Error messages for failed operations

• CRITICAL - Critical errors that may cause shutdown

Examples

Development logging:

logging = LoggingConfig(
    level="DEBUG",
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)

Production logging:

logging = LoggingConfig(
    level="INFO",
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

__init__(level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO', format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s', stderr_output: bool = True) → None

level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']

Logging level. Must be DEBUG, INFO, WARNING, ERROR, or CRITICAL.

format: str

Log message format string. Python logging format.

stderr_output: bool

Log to stderr. Required for MCP protocol compliance.

class rmcp.config.models.RMCPConfig(http: ~rmcp.config.models.HTTPConfig = <factory>, r: ~rmcp.config.models.RConfig = <factory>, security: ~rmcp.config.models.SecurityConfig = <factory>, performance: ~rmcp.config.models.PerformanceConfig = <factory>, logging: ~rmcp.config.models.LoggingConfig = <factory>, config_file: ~pathlib.Path | None = None, debug: bool = False)

Bases: object

Main RMCP configuration.

Root configuration object containing all subsystem configurations. Provides validation and hierarchical configuration loading.

Environment Variables:

• RMCP_DEBUG - Enable debug mode (true/false)

Configuration Loading:

1. Command-line arguments (highest priority)

2. Environment variables (RMCP_* prefix)

3. User config file (~/.rmcp/config.json)

4. System config file (/etc/rmcp/config.json)

5. Built-in defaults (lowest priority)

Examples

Complete configuration:

config = RMCPConfig(
    http=HTTPConfig(host="0.0.0.0", port=8000),
    r=RConfig(timeout=180, max_sessions=20),
    security=SecurityConfig(vfs_read_only=True),
    debug=True
)

Minimal configuration (uses defaults):

config = RMCPConfig(debug=True)

__init__(http: ~rmcp.config.models.HTTPConfig = <factory>, r: ~rmcp.config.models.RConfig = <factory>, security: ~rmcp.config.models.SecurityConfig = <factory>, performance: ~rmcp.config.models.PerformanceConfig = <factory>, logging: ~rmcp.config.models.LoggingConfig = <factory>, config_file: ~pathlib.Path | None = None, debug: bool = False) → None

http: HTTPConfig

HTTP transport configuration.

r: RConfig

R process configuration.

security: SecurityConfig

Security and VFS configuration.

performance: PerformanceConfig

Performance and resource configuration.

logging: LoggingConfig

Logging configuration.

config_file: Path | None

Path to configuration file that was loaded.

debug: bool

Enable debug mode for verbose logging and configuration details.

Configuration Loader

Configuration loading and management for RMCP.

This module implements hierarchical configuration loading with support for multiple configuration sources in priority order:

1. Command-line arguments (highest priority)

2. Environment variables (RMCP_* prefix)

3. User configuration file (~/.rmcp/config.json)

4. System configuration file (/etc/rmcp/config.json)

5. Built-in defaults (lowest priority)

Features:

• Automatic environment variable mapping with RMCP_* prefix

• JSON configuration file validation with schema

• Type conversion and validation

• Configuration caching for performance

• Detailed error reporting with helpful messages

Environment Variable Mapping:

All configuration options can be set via environment variables:

• RMCP_HTTP_HOST → http.host

• RMCP_HTTP_PORT → http.port

• RMCP_R_TIMEOUT → r.timeout

• RMCP_LOG_LEVEL → logging.level

• RMCP_DEBUG → debug

Configuration File Format:

JSON files with nested structure matching the configuration model:

{
  "http": {"host": "0.0.0.0", "port": 8000},
  "r": {"timeout": 180, "max_sessions": 20},
  "security": {"vfs_read_only": true},
  "logging": {"level": "DEBUG"},
  "debug": true
}

Examples

Load configuration with custom file:

loader = ConfigLoader()
config = loader.load_config(config_file="/path/to/config.json")

Load with environment variables:

os.environ["RMCP_HTTP_PORT"] = "9000"
os.environ["RMCP_DEBUG"] = "true"
config = loader.load_config()

Load with CLI overrides:

config = loader.load_config(
    cli_overrides={"debug": True, "http": {"host": "0.0.0.0"}}
)

exception rmcp.config.loader.ConfigError

Bases: Exception

Configuration-related errors.

Raised when configuration loading, validation, or parsing fails. Provides detailed error messages to help users fix configuration issues.

class rmcp.config.loader.ConfigLoader

Bases: object

Handles loading and merging configuration from multiple sources.

The ConfigLoader implements the hierarchical configuration system for RMCP, automatically discovering and merging configuration from multiple sources in priority order.

Features:

• Configuration caching for performance

• Automatic environment variable discovery

• JSON schema validation

• Detailed error reporting

• Type conversion and validation

Usage:

The loader is typically used as a singleton to ensure consistent configuration across the application:

loader = ConfigLoader()
config = loader.load_config()

For custom configuration scenarios:

config = loader.load_config(
    config_file="/custom/path/config.json",
    cli_overrides={"debug": True}
)

__init__()

Initialize the configuration loader.

Creates a new configuration loader with empty cache. The cache will be populated on first load_config() call.

load_config(config_file: str | Path | None = None, overrides: dict[str, Any] | None = None, validate: bool = True) → RMCPConfig

Load configuration from all sources in priority order: 1. Overrides (highest priority) 2. Environment variables 3. Specified config file or auto-discovered files 4. Defaults (lowest priority)

Parameters:

• config_file (str | Path | None) – Explicit config file path

• overrides (dict[str, Any] | None) – Dictionary of override values

• validate (bool) – Whether to validate against JSON schema

Returns:

Loaded and validated RMCPConfig instance

Return type:

RMCPConfig

rmcp.config.loader.get_config(reload: bool = False) → RMCPConfig

Get the global RMCP configuration instance.

Parameters:

reload (bool) – Force reload configuration from sources

Returns:

Global RMCPConfig instance

Return type:

RMCPConfig

rmcp.config.loader.load_config(config_file: str | Path | None = None, overrides: dict[str, Any] | None = None, validate: bool = True) → RMCPConfig

Load a new configuration instance (does not affect global config).

Parameters:

• config_file (str | Path | None) – Path to configuration file

• overrides (dict[str, Any] | None) – Configuration overrides

• validate (bool) – Whether to validate configuration

Returns:

New RMCPConfig instance

Return type:

RMCPConfig

Transport Layer

Base Transport

Base transport interface for MCP server. Defines the contract that all transports must implement, enabling clean composition at the server edge.

class rmcp.transport.base.Transport(name: str)

Bases: ABC

Abstract base class for MCP transports. All transports must implement: - Message receiving (async iterator) - Message sending - Lifecycle management (startup/shutdown) - Error handling

__init__(name: str)

set_message_handler(handler: Callable[[dict[str, Any]], Awaitable[dict[str, Any] | None]]) → None

Set the message handler that will process incoming messages.

abstractmethod async startup() → None

Initialize the transport.

abstractmethod async shutdown() → None

Clean up the transport.

abstractmethod receive_messages() → AsyncIterator[dict[str, Any]]

Async iterator that yields incoming messages. Messages are already parsed from transport format (JSON-RPC).

abstractmethod async send_message(message: dict[str, Any]) → None

Send a message via the transport. Message will be encoded to transport format (JSON-RPC).

async run() → None

Run the transport event loop. This is the main entry point that: 1. Starts the transport 2. Processes incoming messages 3. Handles errors gracefully 4. Ensures clean shutdown

property is_running: bool

Check if transport is running.

HTTP Transport

The HTTP transport module requires optional HTTP dependencies (fastapi, uvicorn). Install with: pip install rmcp[http]

For HTTP API documentation, see HTTP Server API.

Registries

Tool Registry

Tools registry for MCP server. Provides: - @tool decorator for declarative tool registration - Schema validation with proper error codes - Tool discovery and dispatch - Context-aware execution Following the principle: “Registries are discoverable and testable.”

class rmcp.registries.tools.ToolHandler(*args, **kwargs)

Bases: Protocol

Protocol for tool handler functions with MCP metadata.

__init__(*args, **kwargs)

class rmcp.registries.tools.ToolDefinition(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)

Bases: object

Tool metadata and handler.

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

__init__(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

class rmcp.registries.tools.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)

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.

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

List available tools for MCP tools/list.

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

Call a tool with validation.

rmcp.registries.tools.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.registries.tools.register_tool_functions(registry: ToolsRegistry, *functions: ToolHandler) → None

Register multiple functions decorated with @tool.

Resource Registry

Resources registry for MCP server. Implements mature MCP patterns: - Read-only endpoints for files and in-memory objects - URI-based addressing (file://, mem://) - Resource templates for parameterized access - VFS integration for security Following the principle: “Keeps data access explicit and auditable.”

class rmcp.registries.resources.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)

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.

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.

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.

rmcp.registries.resources.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”