Servers
This module contains MCP server implementations.
Jupyter QCodes Server
FastMCP server implementation for Jupyter QCoDeS integration.
This server provides read-only access to QCoDeS instruments and Jupyter notebook functionality through MCP tools.
- class instrmcp.servers.jupyter_qcodes.mcp_server.JupyterMCPServer(ipython, host='127.0.0.1', port=8123, safe_mode=True, dangerous_mode=False, enabled_options=None)[source]
Bases:
objectMCP server for Jupyter QCoDeS integration.
- __init__(ipython, host='127.0.0.1', port=8123, safe_mode=True, dangerous_mode=False, enabled_options=None)[source]
- start_sync()[source]
Synchronous start - works from any context (including after %gui qt).
This is the primary method for starting the server. It creates a dedicated background thread for uvicorn, making it immune to event loop changes in the main thread.
- stop_sync()[source]
Synchronous stop - works from any context (including after %gui qt).
This is the primary method for stopping the server. It signals uvicorn to exit, waits for the thread to finish, and cleans up resources.
- Return type:
- Returns:
True if server stopped successfully, False if timeout occurred.
- is_running()[source]
Thread-safe running check.
Uses our own _server_started flag rather than uvicorn internals. This avoids dependency on uvicorn’s internal API which may change.
- Return type:
- async start()[source]
Start the MCP server (async wrapper for start_sync).
This async method is kept for backward compatibility but internally uses the synchronous thread-based approach.
- async stop()[source]
Stop the MCP server (async wrapper for stop_sync).
This async method is kept for backward compatibility but internally uses the synchronous thread-based approach. Cleanup is handled by stop_sync() so we don’t duplicate it here.
QCodes Tools
Read-only QCoDeS tools for the Jupyter MCP server.
This module provides the QCodesReadOnlyTools facade class that delegates to domain-specific backends for QCodes, notebook, and MeasureIt operations.
The facade maintains backward compatibility while the actual implementation is split across: - backend/qcodes.py: QCodes instrument operations - backend/notebook.py: Read-only notebook operations - backend/notebook_unsafe.py: Unsafe notebook operations (modification, execution) - options/measureit/backend.py: MeasureIt sweep operations (optional)
- class instrmcp.servers.jupyter_qcodes.tools.QCodesReadOnlyTools(ipython, min_interval_s=0.2)[source]
Bases:
objectFacade for read-only QCoDeS instruments and Jupyter integration.
This class delegates to domain-specific backends while maintaining backward compatibility with the original monolithic interface.
- Parameters:
min_interval_s (
float)
- __init__(ipython, min_interval_s=0.2)[source]
Initialize the tools facade.
- Parameters:
ipython – IPython instance for accessing notebook namespace
min_interval_s (
float) – Minimum interval between hardware reads (rate limiting)
- property measureit_backend
Lazy-load MeasureIt backend when first accessed.
- async instrument_info(name, with_values=False, max_depth=4)[source]
Get detailed information about an instrument.
- async get_parameter_info(instrument_name, parameter_name, detailed=False)[source]
Get metadata information about a specific parameter.
- async get_parameter_values(queries)[source]
Get parameter values - supports both single parameter and batch queries.
- async get_editing_cell(fresh_ms=None, line_start=None, line_end=None, max_lines=200)[source]
Get the currently editing cell content from JupyterLab frontend.
- async execute_editing_cell(timeout=30.0)[source]
Execute the currently editing cell and wait for output.
- async add_new_cell(cell_type='code', position='below', content='')[source]
Add a new cell in the notebook.
- async delete_cells_by_number(cell_numbers)[source]
Delete multiple cells by their execution count numbers.
- async wait_for_sweep(var_name, timeout=None, kill=True, progress_callback=None)[source]
Wait for a measureit sweep to finish.
- async wait_for_all_sweeps(timeout=None, kill=True, progress_callback=None)[source]
Wait until all running measureit sweeps finish.
Unsafe Tools
Active Cell Bridge
Active Cell Bridge for Jupyter MCP Extension
Handles communication between JupyterLab frontend and kernel to capture the currently editing cell content via Jupyter comm protocol.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.register_comm_target()[source]
Register the comm target with IPython kernel.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.request_frontend_snapshot()[source]
Request fresh snapshot from current kernel’s frontend.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_active_cell(fresh_ms=None, timeout_s=0.3)[source]
Get the most recent active cell snapshot.
- Parameters:
- Return type:
- Returns:
Dictionary with cell information or None if no data available. Includes metadata fields: - stale (bool): Whether the data is stale - source (str): “live” or “cache” - age_ms (float): Age of the snapshot in milliseconds - stale_reason (str, optional): Reason for staleness if stale=True
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_bridge_status()[source]
Get status information about the bridge.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.update_active_cell(content, timeout_s=2.0)[source]
Update the content of the currently active cell in JupyterLab frontend.
FIX: Now sends to only the current kernel’s comm instead of all comms.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.execute_active_cell(timeout_s=5.0)[source]
Execute the currently active cell in JupyterLab frontend.
FIX: Now sends to only the current kernel’s comm instead of all comms.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.add_new_cell(cell_type='code', position='below', content='', timeout_s=2.0)[source]
Add a new cell relative to the currently active cell in JupyterLab frontend.
FIX: Now sends to only the current kernel’s comm instead of all comms and waits for the frontend response to report actual success/failure.
- Parameters:
cell_type (
str) – Type of cell to create (“code”, “markdown”, “raw”)position (
str) – Position relative to active cell (“above”, “below”, “end”) “end” appends the cell at the very end of the notebookcontent (
str) – Initial content for the new celltimeout_s (
float) – How long to wait for response from frontend (default 2.0s)
- Return type:
- Returns:
Dictionary with creation status and response details
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.delete_editing_cell(timeout_s=2.0)[source]
Delete the currently active cell in JupyterLab frontend.
FIX: Now sends to only the current kernel’s comm instead of all comms. This is the primary fix for Bug #1 - previously this function would send delete requests to ALL connected frontends, causing 2-5 cells to be deleted.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.apply_patch(old_text, new_text, timeout_s=2.0)[source]
Apply a simple text replacement patch to the currently active cell.
FIX: Now sends to only the current kernel’s comm instead of all comms. This is the primary fix for Bug #2 - previously this function would send patch requests to ALL connected frontends, causing the patch to apply multiple times (e.g., “y = 2” -> “y = 200” became “y = 2000000”).
This function replaces the first occurrence of old_text with new_text in the active cell content.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.delete_cells_by_number(cell_numbers, timeout_s=2.0)[source]
Delete multiple cells by their execution count numbers.
FIX: Now sends to only the current kernel’s comm instead of all comms.
This function sends a request to the JupyterLab frontend to delete cells identified by their execution counts.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_cached_cell_output(cell_number, max_age_seconds=None)[source]
Get cached output for a specific cell from the frontend response cache.
Implements timestamp-based cache validation to avoid returning stale error states that no longer reflect the current cell state.
- Parameters:
- Return type:
- Returns:
Dictionary with output data if available and not expired, None otherwise. The returned dict includes metadata: - “data”: The actual output data - “timestamp”: When the data was cached - “age_seconds”: How old the cached data is - “stale”: Whether the data exceeds max_age_seconds
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.invalidate_cell_output_cache(cell_numbers=None, older_than_seconds=None)[source]
Invalidate (clear) cached cell outputs.
This function helps prevent stale error states from persisting in the cache. It can be called after cell re-execution to ensure fresh data is fetched.
- Parameters:
- Returns:
“invalidated_count”: Number of cache entries removed
”remaining_count”: Number of entries still in cache
”cell_numbers”: List of cell numbers that were invalidated
- Return type:
Dictionary with invalidation results
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_cell_outputs(cell_numbers, timeout_s=2.0)[source]
Get outputs for specific cells from the JupyterLab frontend.
FIX: Now sends to only the current kernel’s comm instead of all comms.
Retrieves cell outputs (stdout, stderr, execute_result, errors) from the notebook model in the JupyterLab frontend.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.move_cursor(target, timeout_s=2.0)[source]
Move cursor to a different cell in the notebook.
Waits for frontend response to return actual success/failure status, including errors when the target cell does not exist.
- Parameters:
target (
str) – Where to move the cursor: - “above”: Move to cell above current - “below”: Move to cell below current - “bottom”: Move to the last cell in the notebook (by file order) - “index:N”: Move to cell at position N (0-indexed) - works for ALL cellstimeout_s (
float) – How long to wait for response from frontend (default 2.0s)
- Return type:
- Returns:
Dictionary with operation status, old index, and new index. Returns success=False with error message if target cell not found.
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_active_cell_output(timeout_s=10.0)[source]
Get the output of the currently active cell directly from JupyterLab frontend.
This function retrieves the output from the cell that is currently selected in JupyterLab, avoiding stale state issues with IPython’s In/Out history.
FIX for Bug #10: The previous implementation used IPython’s sys.last_value and Out history which can be stale. This directly queries the JupyterLab frontend for the active cell’s current outputs.
- Parameters:
timeout_s (
float) – How long to wait for response from frontend (default 10.0s)- Returns:
success (bool): Whether the operation succeeded
cell_type (str): Type of the active cell (“code”, “markdown”, etc.)
cell_index (int): Index of the active cell in the notebook
execution_count (int|None): Execution count for code cells
has_output (bool): Whether the cell has any output
has_error (bool): Whether the cell output contains an error
outputs (list): List of output objects (stream, execute_result, error, etc.)
image_paths (list): File paths of saved images (if any)
message (str): Status message
- Return type:
Dictionary with
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_notebook_structure(timeout_s=2.0)[source]
Get lightweight notebook structure (metadata only, no source code).
This function retrieves the list of all cells in the notebook with their metadata (type, position, execution count) but WITHOUT source code, making it fast for large notebooks.
- Parameters:
timeout_s (
float) – How long to wait for response from frontend (default 2.0s)- Returns:
success (bool): Whether the operation succeeded
total_cells (int): Total number of cells in notebook
active_cell_index (int): Index of currently active cell
- cells (list): List of cell metadata dicts with:
cell_id_notebook (int): Position in notebook (0-indexed)
cell_type (str): “code”, “markdown”, or “raw”
cell_execution_number (int|None): IPython counter or null
error (str): Error message if failed
- Return type:
Dictionary with
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.get_cells_by_index(cell_id_notebooks, timeout_s=2.0)[source]
Get specific cells by position index (with source code).
This function fetches specific cells from the notebook by their position, allowing targeted retrieval of cell content without fetching the entire notebook.
- Parameters:
- Returns:
success (bool): Whether the operation succeeded
- cells (list): List of cell dicts with:
cell_id_notebook (int): Position in notebook
cell_type (str): “code”, “markdown”, or “raw”
cell_execution_number (int|None): IPython counter or null
source (str): Cell source code
error (str): Error message if failed
- Return type:
Dictionary with
- instrmcp.servers.jupyter_qcodes.active_cell_bridge.delete_cells_by_index(cell_id_notebooks, timeout_s=2.0)[source]
Delete cells by position index (works for ALL cells including unexecuted ones).
This function deletes cells from the notebook by their position, allowing deletion of markdown cells or unexecuted code cells that don’t have execution counts.
IMPORTANT: This function also invalidates the output cache for any deleted cells that had execution counts, preventing stale data issues.
- Parameters:
- Returns:
success (bool): Whether the operation succeeded
deleted_count (int): Number of cells deleted
cleared_count (int): Number of cells cleared (if last cell)
invalidated_exec_counts (list): Execution counts that were invalidated
message (str): Status message
- Return type:
Dictionary with
Tool Registrars
QCodes Tool Registrar
QCodes instrument tool registrar.
Registers tools for interacting with QCodes instruments.
- class instrmcp.servers.jupyter_qcodes.core.qcodes_tools.QCodesToolRegistrar(mcp_server, tools)[source]
Bases:
objectRegisters QCodes instrument tools with the MCP server.
Notebook Tool Registrar
Jupyter notebook tool registrar.
Registers tools for interacting with Jupyter notebook variables and cells.
- class instrmcp.servers.jupyter_qcodes.core.notebook_tools.NotebookToolRegistrar(mcp_server, tools, ipython, safe_mode=True, dangerous_mode=False, enabled_options=None)[source]
Bases:
objectRegisters Jupyter notebook tools with the MCP server.
- __init__(mcp_server, tools, ipython, safe_mode=True, dangerous_mode=False, enabled_options=None)[source]
Initialize the notebook tool registrar.
- Parameters:
mcp_server – FastMCP server instance
tools – QCodesReadOnlyTools instance
ipython – IPython instance for direct notebook access
safe_mode – Whether server is in safe mode (read-only)
dangerous_mode – Whether server is in dangerous mode (auto-approve consents)
enabled_options – Set of enabled optional features (measureit, database, etc.)
Database Tool Registrar
MeasureIt Tool Registrar
Resource Registrar
Resource registrar for MCP server.
Registers all MCP resources (core, MeasureIt templates, and database resources). Resource descriptions are loaded from metadata_baseline.yaml.
- class instrmcp.servers.jupyter_qcodes.core.resources.ResourceRegistrar(mcp_server, tools, enabled_options=None, measureit_module=None, db_module=None, metadata_config=None)[source]
Bases:
objectRegisters all MCP resources with the server.
- __init__(mcp_server, tools, enabled_options=None, measureit_module=None, db_module=None, metadata_config=None)[source]
Initialize the resource registrar.
- Parameters:
mcp_server – FastMCP server instance
tools – QCodesReadOnlyTools instance
enabled_options – Set of enabled optional features
measureit_module – MeasureIt integration module (optional)
db_module – Database integration module (optional)
metadata_config – MetadataConfig instance for resource descriptions
Cache
Caching and rate limiting for QCoDeS parameter reads.
- class instrmcp.servers.jupyter_qcodes.cache.ReadCache[source]
Bases:
objectThread-safe cache for QCoDeS parameter values with timestamps.
- class instrmcp.servers.jupyter_qcodes.cache.RateLimiter(min_interval_s=0.2)[source]
Bases:
objectRate limiter for QCoDeS instrument access.
- Parameters:
min_interval_s (
float)
- get_instrument_lock(instrument_name)[source]
Get or create a lock for an instrument.
- Parameters:
instrument_name (
str)- Return type:
Lock
- class instrmcp.servers.jupyter_qcodes.cache.ParameterPoller(cache, rate_limiter)[source]
Bases:
objectBackground poller for subscribed parameters.
- Parameters:
cache (
ReadCache)rate_limiter (
RateLimiter)
- __init__(cache, rate_limiter)[source]
- Parameters:
cache (
ReadCache)rate_limiter (
RateLimiter)
- async subscribe(instrument_name, parameter_name, interval_s, get_parameter_func)[source]
Subscribe to periodic parameter updates.
Jupyter Extension
IPython extension entry point for the Jupyter QCoDeS MCP server.
This extension is automatically loaded when installing instrmcp. Manual loading: %load_ext instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension
- class instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.MCPMagics(shell=None, **kwargs)[source]
Bases:
MagicsMagic commands for MCP server control.
- Parameters:
shell (InteractiveShell | None)
kwargs (Any)
- mcp_start(line)[source]
Start the MCP server.
This uses synchronous server start, which works from any context including after %gui qt.
- mcp_close(line)[source]
Stop the MCP server.
This uses synchronous server stop, which works from any context including after %gui qt.
- mcp_restart(line)[source]
Restart the MCP server to apply mode changes.
This uses synchronous server restart, which works from any context including after %gui qt.
- magics = {'cell': {}, 'line': {'mcp_close': 'mcp_close', 'mcp_dangerous': 'mcp_dangerous', 'mcp_option': 'mcp_option', 'mcp_restart': 'mcp_restart', 'mcp_safe': 'mcp_safe', 'mcp_start': 'mcp_start', 'mcp_status': 'mcp_status', 'mcp_unsafe': 'mcp_unsafe'}}
- registered = True
- instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.load_ipython_extension(ipython)[source]
Load the MCP extension when IPython starts.
- instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.unload_ipython_extension(ipython)[source]
Unload the MCP extension when IPython shuts down.
- instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.get_server()[source]
Get the current MCP server instance.
- Return type:
- instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.get_server_status()[source]
Get server status information.
Returns thread-safe status using the server’s is_running() method.
- Return type:
- instrmcp.servers.jupyter_qcodes.jupyter_mcp_extension.broadcast_server_status(status, details=None)[source]
Broadcast server status to all connected toolbar frontends.
Sends through existing toolbar control comms instead of creating new Comms.
This function handles different execution contexts to avoid deadlocks: - If IO loop is running: schedules sends via call_soon_threadsafe - If IO loop exists but not running: calls sends directly (sync context) - If no IO loop available: skips broadcast with a warning
This approach is safer than using a daemon thread, as Jupyter comms are not thread-safe.