Nexus transport for workflow callers #1
Conversation
pyproject.toml
Outdated
|
|
||
| [tool.uv.sources] | ||
| temporalio = { git = "https://github.com/temporalio/sdk-python" } | ||
| temporalio = { git = "https://github.com/temporalio/sdk-python", branch = "dan-9997-python-mcp-nexus" } |
There was a problem hiding this comment.
The Temporal SDK needs a small modification: the event loop needs a stub implementation of get_task_factory
There was a problem hiding this comment.
nit: rename to transport.py since this is already in the nexusmcp package.
There was a problem hiding this comment.
Discussed offline. Renamed to workflow_transport.WorkflowTransport, the idea being that this is transport over "workflow protocol" (or "workflow task protocol"), although those terms are not currently standard.
| from contextlib import asynccontextmanager | ||
| from typing import Any, AsyncGenerator | ||
|
|
||
| import anyio |
There was a problem hiding this comment.
I realized anyio wasn't in the project dependencies but it was already in use in tests before your PR. I wonder if there's a linter that can catch that.
There was a problem hiding this comment.
Added it to dependencies
| async for session_message in transport_read: | ||
| request = session_message.message.root | ||
| if not isinstance(request, types.JSONRPCRequest): | ||
| continue |
There was a problem hiding this comment.
Is this intentionally ignored? Can you add a comment saying why?
There was a problem hiding this comment.
Added a comment saying that we ignore e.g. types.JSONRPCNotification
In practice, the test currently gets JSONRPCNotification method='notifications/initialized'
| match request: | ||
| case types.JSONRPCRequest(method="initialize"): | ||
| result = self._handle_initialize( | ||
| types.InitializeRequestParams.model_validate(request.params) |
There was a problem hiding this comment.
Interesting, I would have expected this to already have come validated from the client. I wonder if there's something to learn here for the Nexus client / transport separation.
There was a problem hiding this comment.
You're right, the client does construct those types explicitly, e.g. https://github.com/modelcontextprotocol/python-sdk/blob/d1ac8d68eb2d7ed139bdc2608b8b4e2ec4265be5/src/mcp/client/session.py#L293-L302
We could cast here instead of validating again, but I think I prefer validating again, e.g. so that our transport can be used safely in other contexts.
| await transport_read.aclose() | ||
|
|
||
| def _handle_initialize(self, params: types.InitializeRequestParams) -> types.InitializeResult: | ||
| # TODO: MCPService should implement this |
There was a problem hiding this comment.
Want to open an issue? Or just add it, it should be easy..
|
Thanks @bergundy, comments addressed and added content to README. |
bergundy
left a comment
bergundy
left a comment
There was a problem hiding this comment.
Approved with a couple of comments.
| from pydantic import BaseModel | ||
| from temporalio import workflow | ||
| from temporalio.client import Client | ||
| from temporalio.contrib.pydantic import pydantic_data_converter |
There was a problem hiding this comment.
Ah, I forgot to add that on the handler side, want to do that before merging? Otherwise, I can take that on.
There was a problem hiding this comment.
Sure -- I've added it to all the client connections in the README and tests.
dandavison
force-pushed
the
workflow-caller
branch
from
42fe0cc to
5244fb7
Compare
|
Comments addressed. |
|
temporalio/sdk-python#1031 is merged so this is no longer pointing at a feature branch |
2 participants
This PR makes it possible to make MCP calls from a Temporal workflow to a Nexus MCP server, via a custom transport for the standard MCP SDK client. The typical use case is an AI agent implemented as a Temporal workflow that wants to make use of tools that are themselves backed by Temporal workflows.