Nexus transport for workflow callers

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+*.egg-info/

README.md

@@ -99,11 +99,12 @@ class CalculatorHandler:
         return CalculateResponse(result=result)
 ```
 
-### 4. Create a Nexus endpoint
+### 4. Create a namespace and Nexus endpoint
 
 **Local dev or self hosted deployment**
 
 ```bash
+temporal operator namespace create --namespace my-handler-namespace
 temporal operator nexus endpoint create \
   --name mcp-gateway \
   --target-namespace my-handler-namespace \
@@ -117,13 +118,20 @@ temporal operator nexus endpoint create \
 ### 3. Set Up the Temporal Worker with the Nexus handlers at `worker.py`
 
 ```python
+import asyncio
+
 from temporalio.client import Client
+from temporalio.contrib.pydantic import pydantic_data_converter
 from temporalio.worker import Worker
 from .service_handler import mcp_service_handler, CalculatorHandler
 
 async def main():
     # Connect to Temporal (replace host and namespace as needed).
-    client = await Client.connect("localhost:7233", namespace="my-handler-namespace")
+    client = await Client.connect(
+        "localhost:7233",
+        namespace="my-handler-namespace",
+        data_converter=pydantic_data_converter,
+    )
 
     async with Worker(
         client,
@@ -136,17 +144,26 @@ async def main():
 
 ### 4. Set Up the MCP Gateway
 
+```bash
+temporal operator namespace create --namespace my-caller-namespace
+```
+
 ```python
 import asyncio
 from mcp.server.lowlevel import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
 from temporalio.client import Client
+from temporalio.contrib.pydantic import pydantic_data_converter
 from nexusmcp import InboundGateway
 
 async def main():
     server = Server("nexus-mcp-demo")
     # Connect to Temporal (replace host and namespace as needed).
-    client = await Client.connect("localhost:7233", namespace="my-caller-namespace")
+    client = await Client.connect(
+        "localhost:7233",
+        namespace="my-caller-namespace",
+        data_converter=pydantic_data_converter,
+    )
 
     # Create the MCP gateway
     gateway = InboundGateway(
@@ -163,7 +180,7 @@ if __name__ == "__main__":
     asyncio.run(main())
 ```
 
-### 4. Configure Your MCP Client
+### 5. Configure Your MCP Client
 
 Add to your MCP client configuration (e.g., Claude Desktop):
 
@@ -178,6 +195,57 @@ Add to your MCP client configuration (e.g., Claude Desktop):
 }
 ```
 
+### 6. Make MCP calls from a Temporal Workflow
+
+```python
+import asyncio
+import uuid
+
+from mcp import ClientSession
+from nexusmcp import WorkflowTransport
+from pydantic import BaseModel
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.contrib.pydantic import pydantic_data_converter
+from temporalio.worker import Worker
+
+
+class AgentWorkflowInput(BaseModel):
+    endpoint: str
+
+
+# The workflow must have the sandbox disabled
+@workflow.defn(sandboxed=False)
+class AgentWorkflow:
+    @workflow.run
+    async def run(self, input: AgentWorkflowInput):
+        transport = WorkflowTransport(input.endpoint)
+        async with transport.connect() as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize()
+                list_tools_result = await session.list_tools()
+                print(f"available tools: {list_tools_result}")
+
+
+async def main():
+    client = await Client.connect(
+        "localhost:7233",
+        data_converter=pydantic_data_converter,
+    )
+
+    async with Worker(
+        client,
+        task_queue="agent-workflow",
+        workflows=[AgentWorkflow],
+    ) as worker:
+        await client.execute_workflow(
+            AgentWorkflow.run,
+            AgentWorkflowInput(endpoint="mcp-gateway"),
+            id=str(uuid.uuid4()),
+            task_queue=worker.task_queue,
+        )
+```
+
 ## Usage Examples
 
 ### Tool Filtering

nexusmcp/__init__.py

@@ -4,5 +4,6 @@
     from .inbound_gateway import InboundGateway
     from .service import MCPService
     from .service_handler import MCPServiceHandler, exclude
+    from .workflow_transport import WorkflowTransport
 
-__all__ = ["MCPService", "MCPServiceHandler", "InboundGateway", "exclude"]
+__all__ = ["MCPService", "MCPServiceHandler", "InboundGateway", "exclude", "WorkflowTransport"]

nexusmcp/inbound_gateway.py

@@ -29,7 +29,7 @@ class InboundGateway:
 
     This gateway acts as an adapter between the Model Context Protocol (MCP) server and Temporal Nexus Operations,
     enabling tool calls to be executed reliably through Temporal's workflow engine. It handles both tool listing and
-    tool call requests by delegating them to corresponding a set of Temporal Nexus Services in a given endpoint.
+    tool call requests by delegating them to a corresponding set of Temporal Nexus Services in a given endpoint.
     """
 
     _client: Client
@@ -93,7 +93,7 @@ async def _handle_call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
                 {"a": 5, "b": 3}
             )
         """
-        service, operation = name.split("/", maxsplit=1)
+        service, _, operation = name.partition("/")
         if not service or not operation:
             raise ValueError(f"Invalid tool name: {name}, must be in the format 'service/operation'")
         return await self._client.execute_workflow(

nexusmcp/proxy_workflow.py

@@ -12,7 +12,6 @@
 
 class ToolListInput(BaseModel):
     endpoint: str
-    pass
 
 
 class ToolCallInput(BaseModel):

nexusmcp/workflow_transport.py

@@ -0,0 +1,152 @@
+import asyncio
+from contextlib import asynccontextmanager
+from typing import Any, AsyncGenerator
+
+import anyio
+import mcp.types as types
+import pydantic
+from mcp.shared.message import SessionMessage
+from temporalio import workflow
+
+from .service import MCPService
+
+
+class WorkflowTransport:
+    """
+    An MCP Transport for use in Temporal workflows.
+
+    This class provides a transport that proxies MCP requests from a Temporal Workflow to a Temporal
+    Nexus service. It can be used to make MCP calls via `mcp.ClientSession` from Temporal workflow
+    code.
+
+    Example:
+        ```python async with WorkflowNexusTransport("my-endpoint") as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize() await session.list_tools() await
+                session.call_tool("my-service/my-operation", {"arg": "value"})
+        ```
+    """
+
+    def __init__(
+        self,
+        endpoint: str,
+    ):
+        self.endpoint = endpoint
+
+    @asynccontextmanager
+    async def connect(
+        self,
+    ) -> AsyncGenerator[
+        tuple[
+            anyio.streams.memory.MemoryObjectReceiveStream[SessionMessage],
+            anyio.streams.memory.MemoryObjectSendStream[SessionMessage],
+        ],
+        None,
+    ]:
+        client_write, transport_read = anyio.create_memory_object_stream(0)  # type: ignore[var-annotated]
+        transport_write, client_read = anyio.create_memory_object_stream(0)  # type: ignore[var-annotated]
+
+        async def message_router() -> None:
+            try:
+                async for session_message in transport_read:
+                    request = session_message.message.root
+                    if not isinstance(request, types.JSONRPCRequest):
+                        # Ignore e.g. types.JSONRPCNotification
+                        continue
+                    result: types.Result | types.ErrorData
+                    try:
+                        match request:
+                            case types.JSONRPCRequest(method="initialize"):
+                                result = self._handle_initialize(
+                                    types.InitializeRequestParams.model_validate(request.params)
+                                )
+                            case types.JSONRPCRequest(method="tools/list"):
+                                result = await self._handle_list_tools()
+                            case types.JSONRPCRequest(method="tools/call"):
+                                result = await self._handle_call_tool(
+                                    types.CallToolRequestParams.model_validate(request.params)
+                                )
+                            case _:
+                                result = types.ErrorData(
+                                    code=types.METHOD_NOT_FOUND, message=f"Unknown method: {request.method}"
+                                )
+                    except pydantic.ValidationError as e:
+                        result = types.ErrorData(code=types.INVALID_PARAMS, message=f"Invalid request: {e}")
+
+                    match result:
+                        case types.Result():
+                            response = self._json_rpc_result_response(request, result)
+                        case types.ErrorData():
+                            response = self._json_rpc_error_response(request, result)
+
+                    await transport_write.send(SessionMessage(types.JSONRPCMessage(root=response)))
+
+            except anyio.ClosedResourceError:
+                pass
+            finally:
+                await transport_write.aclose()
+
+        router_task = asyncio.create_task(message_router())
+
+        try:
+            yield client_read, client_write
+        finally:
+            await client_write.aclose()
+            router_task.cancel()
+            try:
+                await router_task
+            except asyncio.CancelledError:
+                pass
+            await transport_read.aclose()
+
+    def _handle_initialize(self, params: types.InitializeRequestParams) -> types.InitializeResult:
+        # TODO: MCPService should implement this
+        return types.InitializeResult(
+            protocolVersion="2024-11-05",
+            capabilities=types.ServerCapabilities(tools=types.ToolsCapability()),
+            serverInfo=types.Implementation(
+                name="nexus-mcp-transport",
+                version="0.1.0",
+            ),
+        )
+
+    async def _handle_list_tools(self) -> types.ListToolsResult:
+        nexus_client = workflow.create_nexus_client(
+            endpoint=self.endpoint,
+            service=MCPService,
+        )
+        tools = await nexus_client.execute_operation(MCPService.list_tools, None)
+        return types.ListToolsResult(tools=tools)
+
+    async def _handle_call_tool(self, params: types.CallToolRequestParams) -> types.CallToolResult:
+        service, _, operation = params.name.partition("/")
+        nexus_client = workflow.create_nexus_client(
+            endpoint=self.endpoint,
+            service=service,
+        )
+        result: Any = await nexus_client.execute_operation(
+            operation,
+            params.arguments or {},
+        )
+        if isinstance(result, dict):
+            return types.CallToolResult(content=[], structuredContent=result)
+        else:
+            return types.CallToolResult(content=[types.TextContent(type="text", text=str(result))])
+
+    def _json_rpc_error_response(self, request: types.JSONRPCRequest, error: types.ErrorData) -> types.JSONRPCResponse:
+        return types.JSONRPCResponse.model_validate(
+            {
+                "jsonrpc": "2.0",
+                "id": request.id,
+                "error": error.model_dump(),
+            }
+        )
+
+    def _json_rpc_result_response(self, request: types.JSONRPCRequest, result: types.Result) -> types.JSONRPCResponse:
+        return types.JSONRPCResponse.model_validate(
+            {
+                "jsonrpc": "2.0",
+                "id": request.id,
+                "result": result.model_dump(),
+            }
+        )

pyproject.toml

@@ -5,6 +5,7 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "anyio>=4.10.0",
     "mcp>=1.13.0",
     "nexus-rpc>=1.1.0",
     "pydantic>=2.11.7",

tests/test_inbound_gateway.py

@@ -1,13 +1,14 @@
 import asyncio
 
 import anyio
-from mcp.shared.message import SessionMessage
 import pytest
 from mcp import ClientSession
 from mcp.server.lowlevel import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
+from mcp.shared.message import SessionMessage
 from temporalio.api.nexus.v1 import EndpointSpec, EndpointTarget
 from temporalio.api.operatorservice.v1 import CreateNexusEndpointRequest
+from temporalio.contrib.pydantic import pydantic_data_converter
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
@@ -23,7 +24,7 @@ async def test_inbound_gateway() -> None:
     endpoint_name = "endpoint"
     task_queue = "handler-queue"
 
-    async with await WorkflowEnvironment.start_local() as env:
+    async with await WorkflowEnvironment.start_local(data_converter=pydantic_data_converter) as env:
         await env.client.operator_service.create_nexus_endpoint(
             CreateNexusEndpointRequest(
                 spec=EndpointSpec(