plexus.cli.procedure.mcp_transport module

In-Process MCP Transport for Procedures

This module provides an in-process implementation of the Model Context Protocol (MCP) that allows experiments to provide MCP tools directly to AI models without requiring a separate server process.

The implementation follows MCP architectural patterns but uses direct method calls instead of stdio or HTTP transport, providing better performance and shared state access for procedure contexts.

class plexus.cli.procedure.mcp_transport.EmbeddedMCPServer(experiment_context: Dict[str, Any] | None = None)

Bases: object

An embedded MCP server that runs within an procedure process. Provides MCP tools and resources without requiring a separate server process.

__init__(experiment_context: Dict[str, Any] | None = None)

connect(client_info: Dict[str, Any] | None = None)

Context manager for MCP connection lifecycle.

Usage:

async with server.connect(client_info) as mcp_client:: tools = await mcp_client.list_tools() result = await mcp_client.call_tool(“tool_name”, {“arg”: “value”})

register_plexus_tools(tool_subset: List[str] | None = None)

Register a subset of Plexus MCP tools for use in experiments.

Args:: tool_subset: List of tool names to register. If None, registers all available tools.

class plexus.cli.procedure.mcp_transport.InProcessMCPTransport

Bases: object

In-process transport for MCP that uses direct method calls instead of stdio or HTTP. Follows MCP JSON-RPC message format but executes synchronously.

__init__()

async call_tool(name: str, arguments: Dict[str, Any]) → Dict[str, Any]: Call a tool (equivalent to MCP tools/call).

async initialize(client_info: Dict[str, Any]) → Dict[str, Any]: Initialize the MCP connection (equivalent to MCP initialize).

async list_resources() → List[Dict[str, Any]]: List available resources (equivalent to MCP resources/list).

async list_tools() → List[Dict[str, Any]]: List available tools (equivalent to MCP tools/list).

async read_resource(uri: str) → Dict[str, Any]: Read a resource (equivalent to MCP resources/read).

register_resource(resource_info: MCPResourceInfo): Register an MCP resource with the transport.

register_tool(tool_info: MCPToolInfo): Register an MCP tool with the transport.

class plexus.cli.procedure.mcp_transport.MCPResourceInfo(uri: str, name: str, description: str, mime_type: str, handler: Callable[[str], str])

Bases: object

Information about an MCP resource.

__init__(uri: str, name: str, description: str, mime_type: str, handler: Callable[[str], str]) → None

description: str

handler: Callable[[str], str]

mime_type: str

name: str

uri: str

class plexus.cli.procedure.mcp_transport.MCPToolInfo(name: str, description: str, input_schema: Dict[str, Any], handler: Callable[[Dict[str, Any]], Dict[str, Any]])

Bases: object

Information about an MCP tool.

__init__(name: str, description: str, input_schema: Dict[str, Any], handler: Callable[[Dict[str, Any]], Dict[str, Any]]) → None

description: str

handler: Callable[[Dict[str, Any]], Dict[str, Any]]

input_schema: Dict[str, Any]

name: str

class plexus.cli.procedure.mcp_transport.ProcedureMCPClient(transport: InProcessMCPTransport)

Bases: object

Client interface for interacting with the embedded MCP server. Provides async methods that match the MCP protocol.

__init__(transport: InProcessMCPTransport)

async call_tool(name: str, arguments: Dict[str, Any]) → Dict[str, Any]: Call an MCP tool with the given arguments.

async list_resources() → List[Dict[str, Any]]: List available MCP resources.

async list_tools() → List[Dict[str, Any]]: List available MCP tools.

async read_resource(uri: str) → Dict[str, Any]: Read an MCP resource by URI.

async plexus.cli.procedure.mcp_transport.create_procedure_mcp_server(experiment_context: Dict[str, Any] | None = None, plexus_tools: List[str] | None = None) → EmbeddedMCPServer

Create and configure an embedded MCP server for procedure use.

Args:: experiment_context: Context information to make available to MCP tools plexus_tools: List of Plexus tool categories to register.

If None, registers all available tools (default behavior).
Returns:: Configured EmbeddedMCPServer instance