Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 33 additions & 17 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }})
Expand Down
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
72 changes: 72 additions & 0 deletions temporal_mcp/README.md
Original file line number Diff line number Diff line change
@@ -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
```
47 changes: 47 additions & 0 deletions temporal_mcp/pyproject.toml
Original file line number Diff line number Diff line change
@@ -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
65 changes: 65 additions & 0 deletions temporal_mcp/run_worker.py
Original file line number Diff line number Diff line change
@@ -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))
32 changes: 32 additions & 0 deletions temporal_mcp/run_workflow.py
Original file line number Diff line number Diff line change
@@ -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)))
61 changes: 61 additions & 0 deletions temporal_mcp/server.py
Original file line number Diff line number Diff line change
@@ -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()
Loading
Loading