Stub the nexus service

Author: dandavison

mcp_examples/common/mcp_server_nexus_service.py

@@ -1,9 +1,8 @@
 """
 A Nexus service that presents the interface of an MCP server.
-It is backed by a Temporal workflow.
+Returns mock data for testing the transport layer.
 """
 
-import uuid
 from dataclasses import dataclass
 
 import nexusrpc
@@ -12,8 +11,9 @@
     CallToolResult,
     ListToolsRequest,
     ListToolsResult,
+    TextContent,
+    Tool,
 )
-from temporalio import nexus
 
 
 @dataclass
@@ -42,29 +42,72 @@ class MCPServerNexusService:
 
 @nexusrpc.handler.service_handler(service=MCPServerNexusService)
 class MCPServerNexusServiceHandler:
-    @nexus.workflow_run_operation
+    @nexusrpc.handler.sync_operation
     async def start(
-        self, ctx: nexus.WorkflowRunOperationContext, input: MCPServerStartInput
-    ) -> nexus.WorkflowHandle[None]:
-        return await ctx.start_workflow(
-            input.mcp_server_workflow_name,
-            id=str(uuid.uuid4()),
-        )
+        self, ctx: nexusrpc.handler.StartOperationContext, input: MCPServerStartInput
+    ) -> None:
+        # Mock implementation - just return None
+        return None
 
     @nexusrpc.handler.sync_operation
     async def call_tool(
         self, ctx: nexusrpc.handler.StartOperationContext, input: CallToolInput
     ) -> CallToolResult:
-        workflow_handle = nexus.WorkflowHandle.from_token(
-            input.operation_token
-        )._to_client_workflow_handle(nexus.client())
-        return await workflow_handle.execute_update("call_tool", input.request)
+        # Mock implementation - return a sample response
+        if input.request.params.name == "sequentialthinking":
+            args = input.request.params.arguments or {}
+            thought = (
+                args.get("thought", "Mock thought")
+                if isinstance(args, dict)
+                else "Mock thought"
+            )
+            thought_number = (
+                args.get("thoughtNumber", 1) if isinstance(args, dict) else 1
+            )
+            total_thoughts = (
+                args.get("totalThoughts", 3) if isinstance(args, dict) else 3
+            )
+
+            response_text = f"Mock response: Processed thought {thought_number}/{total_thoughts}: {thought}"
+
+            return CallToolResult(
+                content=[TextContent(type="text", text=response_text)]
+            )
+        else:
+            return CallToolResult(
+                content=[
+                    TextContent(
+                        type="text",
+                        text=f"Mock response: Tool {input.request.params.name} called",
+                    )
+                ]
+            )
 
     @nexusrpc.handler.sync_operation
     async def list_tools(
         self, ctx: nexusrpc.handler.StartOperationContext, input: ListToolsInput
     ) -> ListToolsResult:
-        workflow_handle = nexus.WorkflowHandle.from_token(
-            input.operation_token
-        )._to_client_workflow_handle(nexus.client())
-        return await workflow_handle.query("list_tools", input.request)
+        # Mock implementation - return sequential thinking tool
+        return ListToolsResult(
+            tools=[
+                Tool(
+                    name="sequentialthinking",
+                    description="A tool for sequential thinking and problem-solving",
+                    inputSchema={
+                        "type": "object",
+                        "properties": {
+                            "thought": {"type": "string"},
+                            "thoughtNumber": {"type": "integer"},
+                            "totalThoughts": {"type": "integer"},
+                            "nextThoughtNeeded": {"type": "boolean"},
+                        },
+                        "required": [
+                            "thought",
+                            "thoughtNumber",
+                            "totalThoughts",
+                            "nextThoughtNeeded",
+                        ],
+                    },
+                )
+            ]
+        )

mcp_examples/nexus_transport/app.py

@@ -7,15 +7,14 @@
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
 from mcp.types import TextContent
-from temporalio import workflow
-from temporalio.client import Client
-from temporalio.worker import UnsandboxedWorkflowRunner, Worker
-
 from mcp_examples.common.mcp_sdk_nexus_transport import NexusTransport
 from mcp_examples.common.mcp_server_nexus_service import (
     MCPServerNexusService,
     MCPServerNexusServiceHandler,
 )
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.worker import UnsandboxedWorkflowRunner, Worker
 
 
 async def create_client_session_and_call_tool_using_standard_transport():
@@ -32,11 +31,16 @@ async def create_client_session_and_call_tool_using_standard_transport():
 async def create_client_session_and_call_tool_using_nexus_transport(
     nexus_client: workflow.NexusClient[MCPServerNexusService],
 ):
+    print("Creating NexusTransport...")
     transport = NexusTransport(nexus_client, "mcp-sequential-thinking-nexus-endpoint")
 
+    print("Connecting transport...")
     async with transport.connect() as (read_stream, write_stream):
+        print("Creating ClientSession...")
         async with ClientSession(read_stream, write_stream) as session:
+            print("Initializing session...")
             await session.initialize()
+            print("Calling tool...")
             await _call_tool(session)
 
 
@@ -52,13 +56,17 @@ def __init__(self):
 
     @workflow.run
     async def run(self):
+        print("CallToolWorkflow.run() started")
         await create_client_session_and_call_tool_using_nexus_transport(
             self.nexus_client
         )
+        print("CallToolWorkflow.run() completed")
 
 
 async def main():
+    print("Connecting to Temporal...")
     client = await Client.connect("localhost:7233")
+    print("Creating worker...")
     async with Worker(
         client,
         task_queue="mcp-sequential-thinking-task-queue",
@@ -68,11 +76,13 @@ async def main():
         nexus_service_handlers=[MCPServerNexusServiceHandler()],
         workflow_runner=UnsandboxedWorkflowRunner(),
     ) as worker:
-        await client.execute_workflow(
+        print("Executing workflow...")
+        result = await client.execute_workflow(
             CallToolWorkflow.run,
             id=str(uuid.uuid4()),
             task_queue=worker.task_queue,
         )
+        print(f"Workflow completed: {result}")
 
 
 async def _call_tool(session: ClientSession):