Temporal Nexus Calculator MCP Server

# Nexus MCP Calculator - Developer Guide ## Project Overview This is a comprehensive sample implementation of a calculator service that bridges MCP (Model Context Protocol) with Temporal's Nexus RPC framework. It demonstrates the complete development workflow for creating MCP-enabled tools backed by Temporal's reliable execution engine. ## Architecture Components ### Core Service Layer #### `service.py` Contains all Pydantic model definitions and the main `CalculatorService` class: - **Request Models**: `CalculateRequest`, `AddRequest`, `SubtractRequest`, etc. - **Response Models**: `CalculateResponse`, `BasicOperationResponse` - **Service Definition**: `@nexusrpc.service` decorated class with operation definitions - **Type Safety**: All operations use strict type hints and Pydantic validation #### `service_handler.py` The Nexus operation layer that starts workflows for business logic: - **`CalculatorHandler`**: Nexus service handler with `@mcp_service_handler.register` decorator - **Workflow Operations**: Uses `@nexus.workflow_run_operation` to start workflows for each operation - **MCP Integration**: Automatically exposes operations as MCP tools via the registry - **Temporal Visibility**: Each operation creates visible workflow executions in the Temporal UI #### `workflows.py` Temporal workflows that orchestrate calculator operations: - **Workflow Definitions**: One workflow class per calculator operation (CalculateWorkflow, AddWorkflow, etc.) - **Activity Execution**: Each workflow executes one corresponding activity for the actual calculation - **Temporal Benefits**: Workflows provide durability, observability, and fault tolerance - **UI Visibility**: All workflow executions appear in the Temporal UI for monitoring and debugging #### `activities.py` Temporal activities that perform the actual calculations: - **`SafeExpressionEvaluator`**: AST-based mathematical expression parser for security (moved from service_handler) - **Activity Definitions**: One activity per calculator operation with `@activity.defn` decorators - **Business Logic**: Contains the actual mathematical computation logic - **Error Handling**: Comprehensive error handling with activity-specific logging ### Infrastructure Layer #### `worker.py` Temporal worker that runs calculator service handlers, workflows, and activities: - **`CalculatorWorker`**: Worker class with connection management - **Service Registration**: Registers `CalculatorHandler` and `mcp_service_handler` - **Workflow Registration**: Registers all calculator workflow classes - **Activity Registration**: Registers all calculator activity functions - **Configuration**: Command-line arguments for host, namespace, task queue - **Monitoring**: Comprehensive logging with workflow and activity status reporting #### `mcp_server.py` MCP server that bridges clients to Temporal: - **`CalculatorMCPServer`**: MCP server implementation - **Gateway Setup**: Creates and configures `InboundGateway` - **Transport**: Uses stdio transport for MCP communication - **Tool Discovery**: Automatically discovers and exposes Nexus operations #### `demo_workflow.py` Demonstration workflow showing MCP usage from Temporal: - **`CalculatorDemoWorkflow`**: Unsandboxed workflow that calls MCP tools - **Tool Discovery**: Shows how workflows can discover available MCP tools - **Operation Calls**: Demonstrates various calculator operations - **Result Handling**: Proper parsing and aggregation of results ## Key Implementation Patterns ### Security-First Expression Evaluation ```python class SafeExpressionEvaluator: """Uses AST parsing to safely evaluate mathematical expressions.""" ALLOWED_OPERATORS = { ast.Add: operator.add, ast.Sub: operator.sub, # ... only mathematical operators } ALLOWED_FUNCTIONS = { "abs": abs, "round": round, # ... only safe functions } ``` This prevents code injection while allowing complex mathematical expressions. ### MCP Service Registration Pattern ```python mcp_service_handler = MCPServiceHandler() @mcp_service_handler.register @nexusrpc.handler.service_handler(service=CalculatorService) class CalculatorHandler: # All operations automatically become MCP tools ``` The registration pattern automatically exposes Nexus operations as MCP tools with proper metadata. ### Workflow-Activity Pattern The new architecture uses a three-tier approach for maximum observability: ```python # 1. Nexus Operation (service_handler.py) - starts workflow @nexus.workflow_run_operation async def calculate(self, ctx: nexus.WorkflowRunOperationContext, input: CalculateRequest): return await ctx.start_workflow( CalculateWorkflow.run, input, id=f"calculate-{uuid.uuid4()}", ) # 2. Workflow (workflows.py) - orchestrates activity execution @workflow.defn class CalculateWorkflow: @workflow.run async def run(self, input: CalculateRequest) -> CalculateResponse: return await workflow.execute_activity( calculate_activity, input, start_to_close_timeout=workflow.timedelta(seconds=30), ) # 3. Activity (activities.py) - performs actual calculation @activity.defn async def calculate_activity(input: CalculateRequest) -> CalculateResponse: result = _evaluator.evaluate(input.expression) return CalculateResponse(result=result, expression=input.expression) ``` This pattern provides full visibility in the Temporal UI with both workflow and activity executions. ### Error Handling Pattern ```python # In activities - raise standard exceptions that Temporal handles @activity.defn async def divide_activity(input: DivideRequest) -> BasicOperationResponse: if input.b == 0: raise ValueError("Division by zero is not allowed") # Temporal automatically retries and provides stack traces ``` Activities use standard Python exceptions while Temporal provides the reliability layer. ### Workflow Transport Pattern ```python # In workflows, use WorkflowTransport to call MCP tools transport = WorkflowTransport(endpoint_name) async with transport.connect() as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: result = await session.call_tool(name="calculate", arguments={...}) ``` This allows workflows to consume MCP tools through Nexus endpoints. ## Development Workflow ### 1. Service Definition Define your service operations with proper Pydantic models and type hints. ### 2. Handler Implementation Implement the actual business logic with comprehensive error handling. ### 3. Registration Register handlers with the MCP service handler to auto-expose as tools. ### 4. Infrastructure Setup Use the setup scripts to create required Temporal namespaces and endpoints. ### 5. Testing Run the demo workflow to validate end-to-end functionality. ## Testing Strategy ### Unit Testing Test individual calculator operations and expression evaluation: ```python def test_safe_evaluator(): evaluator = SafeExpressionEvaluator() assert evaluator.evaluate("2 + 3 * 4") == 14.0 ``` ### Integration Testing Test MCP tool discovery and execution: ```python async def test_mcp_tools(): # Test that all expected tools are discovered # Test tool execution with various inputs # Test error handling ``` ### End-to-End Testing Use the demo workflow to test the complete system integration. ## Configuration Management ### Environment Variables - `TEMPORAL_HOST`: Temporal server address - `TEMPORAL_NAMESPACE`: Target namespace - `NEXUS_ENDPOINT`: Endpoint name ### Command Line Arguments All components support configurable hosts, namespaces, and endpoints for different environments. ### Setup Scripts Parameterized scripts handle infrastructure creation and cleanup. ## Monitoring and Observability ### Temporal UI Integration - **Workflow Executions**: Every calculator operation creates a visible workflow at http://localhost:8233 - **Activity Executions**: See the actual calculation activities within each workflow - **Execution History**: Full timeline of operation calls with input/output details - **Error Tracking**: Stack traces and retry attempts for failed operations - **Performance Metrics**: Duration, success rates, and throughput statistics - **Nexus Endpoint Health**: Monitor cross-namespace communication - **Worker Status**: Track worker health and task processing rates ### Logging Strategy - Structured logging with appropriate levels - Operation-specific context - Performance metrics ### Error Tracking - Nexus error codes for categorization - Detailed error messages for debugging - Stack traces in development mode ## Extending the Calculator ### Adding New Operations 1. **Define Models** in `service.py`: ```python class NewOpRequest(BaseModel): # Define request structure class NewOpResponse(BaseModel): # Define response structure ``` 2. **Add to Service**: ```python @nexusrpc.service(name="Calculator") class CalculatorService: new_op: nexusrpc.Operation[NewOpRequest, NewOpResponse] ``` 3. **Create Activity** in `activities.py`: ```python @activity.defn async def new_op_activity(input: NewOpRequest) -> NewOpResponse: # Actual calculation logic here return NewOpResponse(...) ``` 4. **Create Workflow** in `workflows.py`: ```python @workflow.defn class NewOpWorkflow: @workflow.run async def run(self, input: NewOpRequest) -> NewOpResponse: return await workflow.execute_activity( new_op_activity, input, start_to_close_timeout=workflow.timedelta(seconds=30), ) ``` 5. **Implement Nexus Operation** in `service_handler.py`: ```python @nexus.workflow_run_operation async def new_op(self, ctx: nexus.WorkflowRunOperationContext, input: NewOpRequest): return await ctx.start_workflow( NewOpWorkflow.run, input, id=f"new_op-{uuid.uuid4()}", ) ``` 6. **Register in Worker** in `worker.py`: ```python # Add to workflows list NewOpWorkflow, # Add to activities list new_op_activity, ``` The operation automatically becomes available as an MCP tool with full Temporal UI visibility. ### Excluding Operations Use the `@exclude` decorator to prevent operations from being exposed: ```python @exclude @nexusrpc.handler.sync_operation async def internal_op(self, _ctx, input): # Not available to MCP clients ``` ## Production Considerations ### Performance - Connection pooling for Temporal clients - Worker scaling based on load - Operation timeout configuration ### Reliability - Circuit breaker patterns for external calls - Retry policies for transient failures - Health check endpoints ### Security - Input validation and sanitization - Rate limiting for operations - Audit logging for sensitive operations ## Dependencies ### Core Dependencies - `nexus-mcp`: The bridge library (local editable install) - `temporalio`: Temporal Python SDK - `nexus-rpc`: Nexus RPC framework - `mcp`: Model Context Protocol implementation - `pydantic`: Type validation and serialization ### Development Dependencies - `pytest`: Testing framework - `mypy`: Static type checking - `ruff`: Code formatting and linting The project uses `uv` for fast, reliable package management and virtual environment handling.