Add Temporal MCP transport samples

Author: brianstrauch

.github/workflows/ci.yml

@@ -25,25 +25,41 @@ jobs:
         env:
           BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}
         run: |
-          if [[ -n "$BASE_SHA" ]] && git cat-file -e "$BASE_SHA^{commit}"; then
-            changed_files="$(git diff --name-only "$BASE_SHA" "$GITHUB_SHA")"
-          else
-            changed_files="$(git ls-files)"
-          fi
+          python - <<'PY'
+          import json
+          import os
+          import subprocess
+          from pathlib import Path
 
-          projects_json="$({
-            for project in \
-              bedrock_agentcore/strands_agent \
-              lambda_worker \
-              openai_agents/mcp_v2
-            do
-              if grep -q "^${project}/" <<< "$changed_files"; then
-                printf '%s\n' "$project"
-              fi
-            done
-          } | jq -Rsc 'split("\n") | map(select(length > 0))')"
+          from scripts.run_root_tool import find_nested_projects
 
-          echo "projects=$projects_json" >> "$GITHUB_OUTPUT"
+          base = os.environ.get("BASE_SHA")
+          if base and subprocess.run(
+              ["git", "cat-file", "-e", f"{base}^{{commit}}"],
+              check=False,
+          ).returncode == 0:
+              changed_output = subprocess.check_output(
+                  ["git", "diff", "--name-only", base, os.environ["GITHUB_SHA"]],
+                  text=True,
+              )
+          else:
+              changed_output = subprocess.check_output(
+                  ["git", "ls-files"],
+                  text=True,
+              )
+
+          changed_files = changed_output.splitlines()
+          projects = [
+              project
+              for project in find_nested_projects(Path.cwd())
+              if any(
+                  path == project or path.startswith(f"{project}/")
+                  for path in changed_files
+              )
+          ]
+          with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+              output.write(f"projects={json.dumps(projects)}\n")
+          PY
 
   standalone-lint-test:
     name: Standalone (${{ matrix.project }}, Python ${{ matrix.python }})

README.md

@@ -110,6 +110,7 @@ without wrapping them in a workflow.
 * [sentry](sentry) - Report errors to Sentry.
 * [sleep_for_days](sleep_for_days) - A workflow that runs forever, sending an email every 30 days.
 * [strands_plugin](strands_plugin) - Run Strands Agents as durable Temporal workflows (model calls, tools, MCP, HITL).
+* [temporal_mcp](temporal_mcp) - Call MCP tools, prompts, and resources durably from Temporal workflows.
 * [trio_async](trio_async) - Use asyncio Temporal in Trio-based environments.
 * [updatable_timer](updatable_timer) - A timer that can be updated while sleeping.
 * [worker_multiprocessing](worker_multiprocessing) - Leverage Python multiprocessing to parallelize workflow tasks and other CPU bound operations by running multiple workers.

temporal_mcp/README.md

@@ -0,0 +1,72 @@
+# Durable MCP clients
+
+This standalone project demonstrates
+[`temporalio-mcp` 0.2.0](https://pypi.org/project/temporalio-mcp/), which lets
+Temporal Workflow code call MCP tools, prompts, and resources through durable
+Activities. The real MCP client, transport, network connections, and credentials
+remain on the Worker.
+
+The same Workflow is available over three MCP SDK v2 client forms:
+
+- `in-process` constructs the MCP server directly inside the Worker process.
+- `stdio` lets the Worker launch and manage `server.py` as a subprocess.
+- `streamable-http` connects the Worker to a separately running HTTP server.
+
+## Prerequisites
+
+- Python 3.10 or newer and [uv](https://docs.astral.sh/uv/)
+- A local Temporal server: `temporal server start-dev`
+
+Change to this directory and install the locked environment:
+
+```bash
+cd temporal_mcp
+uv sync --locked --all-groups
+```
+
+## Run a sample
+
+For the in-process transport, start the Worker and Workflow in separate shells:
+
+```bash
+uv run run_worker.py in-process
+uv run run_workflow.py in-process
+```
+
+The stdio transport uses the same commands. The Worker launches `server.py`
+automatically with the current Python interpreter:
+
+```bash
+uv run run_worker.py stdio
+uv run run_workflow.py stdio
+```
+
+Streamable HTTP needs a third shell for the MCP server:
+
+```bash
+uv run server.py streamable-http
+uv run run_worker.py streamable-http
+uv run run_workflow.py streamable-http
+```
+
+The HTTP server defaults to `http://127.0.0.1:8000/mcp`. Use `--host` and
+`--port` on `server.py` and `--http-url` on `run_worker.py` to change it.
+
+## What the Workflow demonstrates
+
+`MCPDemoWorkflow` lists and calls tools, lists and gets prompts, and lists and
+reads both static and templated resources. Each MCP operation is a Temporal
+Activity. The second `list_tools()` call uses the Workflow client's replay-safe
+cache and does not schedule another Activity.
+
+The sample bounds each Activity attempt and the complete retry series. MCP tools
+still have at-least-once execution semantics, so tools with side effects should
+accept a stable idempotency key. Resolve URLs, tokens, and other secrets in the
+worker-side client factory rather than putting them in Workflow inputs or the
+factory argument, where they would be recorded in Workflow history.
+
+Run all three transports end-to-end against a local Temporal test server with:
+
+```bash
+uv run poe test
+```

temporal_mcp/pyproject.toml

@@ -0,0 +1,47 @@
+[project]
+name = "temporalio-samples-temporal-mcp"
+version = "0.1a1"
+description = "Temporal.io Python SDK durable MCP client samples"
+authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }]
+requires-python = ">=3.10"
+readme = "README.md"
+license = "MIT"
+dependencies = [
+    "temporalio-mcp==0.2.0",
+    "mcp>=2,<3",
+    "mcp-types>=2,<3",
+    "uvicorn>=0.31,<1",
+]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.5.0,<0.6",
+    "mypy>=1.4.1,<2",
+    "poethepoet>=0.36.0",
+    "pytest>=7.1.2,<8",
+    "pytest-asyncio>=0.18.3,<0.19",
+]
+
+[tool.poe.tasks]
+format = [
+    { cmd = "uv run ruff check --select I --fix" },
+    { cmd = "uv run ruff format" },
+]
+lint = [
+    { cmd = "uv run ruff check --select I" },
+    { cmd = "uv run ruff format --check" },
+    { ref = "lint-types" },
+]
+lint-types = "uv run --all-groups mypy --check-untyped-defs --namespace-packages --explicit-package-bases ."
+test = "uv run pytest"
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+pythonpath = ["."]
+testpaths = ["tests"]
+
+[tool.ruff]
+target-version = "py310"
+
+[tool.mypy]
+ignore_missing_imports = true

temporal_mcp/run_worker.py

@@ -0,0 +1,65 @@
+"""Run a Worker configured for one of the sample MCP transports."""
+
+import argparse
+import asyncio
+import sys
+from collections.abc import Callable
+from pathlib import Path
+from typing import cast
+
+from mcp import Client as MCPClient
+from mcp import StdioServerParameters, stdio_client
+from temporalio.client import Client as TemporalClient
+from temporalio.envconfig import ClientConfig
+from temporalio.mcp import MCPPlugin
+from temporalio.worker import Worker
+
+from server import create_server
+from workflow import TRANSPORTS, MCPDemoWorkflow, Transport, client_name, task_queue
+
+SERVER_PATH = Path(__file__).parent / "server.py"
+DEFAULT_HTTP_URL = "http://127.0.0.1:8000/mcp"
+
+
+def client_factory(
+    transport: Transport, http_url: str = DEFAULT_HTTP_URL
+) -> Callable[[], MCPClient]:
+    """Create the worker-side client factory for a transport."""
+    if transport == "in-process":
+        return lambda: MCPClient(create_server())
+    if transport == "stdio":
+        parameters = StdioServerParameters(
+            command=sys.executable,
+            args=[str(SERVER_PATH), "stdio"],
+        )
+        return lambda: MCPClient(stdio_client(parameters))
+    return lambda: MCPClient(http_url)
+
+
+def create_plugin(transport: Transport, http_url: str = DEFAULT_HTTP_URL) -> MCPPlugin:
+    return MCPPlugin({client_name(transport): client_factory(transport, http_url)})
+
+
+async def main(transport: Transport, http_url: str) -> None:
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await TemporalClient.connect(
+        **config,
+        plugins=[create_plugin(transport, http_url)],
+    )
+
+    worker = Worker(
+        client,
+        task_queue=task_queue(transport),
+        workflows=[MCPDemoWorkflow],
+    )
+    print(f"Worker started for {transport}. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("transport", choices=TRANSPORTS)
+    parser.add_argument("--http-url", default=DEFAULT_HTTP_URL)
+    args = parser.parse_args()
+    asyncio.run(main(cast(Transport, args.transport), args.http_url))

temporal_mcp/run_workflow.py

@@ -0,0 +1,32 @@
+"""Start the durable MCP workflow for a selected transport."""
+
+import argparse
+import asyncio
+from typing import cast
+from uuid import uuid4
+
+from temporalio.client import Client
+from temporalio.envconfig import ClientConfig
+
+from workflow import TRANSPORTS, MCPDemoWorkflow, Transport, task_queue
+
+
+async def main(transport: Transport) -> None:
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
+
+    result = await client.execute_workflow(
+        MCPDemoWorkflow.run,
+        transport,
+        id=f"temporal-mcp-{transport}-{uuid4()}",
+        task_queue=task_queue(transport),
+    )
+    print(result)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("transport", choices=TRANSPORTS)
+    args = parser.parse_args()
+    asyncio.run(main(cast(Transport, args.transport)))

temporal_mcp/server.py

@@ -0,0 +1,61 @@
+"""MCP server shared by the in-process, stdio, and HTTP samples."""
+
+import argparse
+from typing import Any
+
+from mcp.server.mcpserver import MCPServer
+
+
+def create_server() -> MCPServer[Any]:
+    """Create the sample server with tools, prompts, and resources."""
+    server = MCPServer("temporal-mcp-sample")
+
+    @server.tool()
+    def echo(value: str) -> str:
+        """Return the supplied value unchanged."""
+        return value
+
+    @server.prompt()
+    def greeting(name: str) -> str:
+        """Create a greeting prompt."""
+        return f"Hello, {name}!"
+
+    @server.resource("sample://about", name="about")
+    def about() -> str:
+        """Describe this sample."""
+        return "Temporal workflows can call MCP operations durably."
+
+    @server.resource("sample://items/{item}", name="item")
+    def item(item: str) -> str:
+        """Return a resource identified by its path parameter."""
+        return item
+
+    return server
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run the sample MCP server")
+    parser.add_argument(
+        "transport",
+        choices=("stdio", "streamable-http"),
+        nargs="?",
+        default="stdio",
+    )
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", default=8000, type=int)
+    args = parser.parse_args()
+
+    server = create_server()
+    if args.transport == "stdio":
+        server.run()
+    else:
+        server.run(
+            transport="streamable-http",
+            host=args.host,
+            port=args.port,
+            stateless_http=True,
+        )
+
+
+if __name__ == "__main__":
+    main()