diff --git a/.github/dependabot.yml b/.github/dependabot.yml deleted file mode 100644 index 62ace76..0000000 --- a/.github/dependabot.yml +++ /dev/null @@ -1,18 +0,0 @@ -version: 2 -updates: - - package-ecosystem: pip - directory: / - schedule: - interval: weekly - open-pull-requests-limit: 5 - labels: - - dependencies - - - package-ecosystem: github-actions - directory: / - schedule: - interval: weekly - open-pull-requests-limit: 5 - labels: - - dependencies - - ci diff --git a/.github/workflows/dependabot-auto-merge.yml b/.github/workflows/dependabot-auto-merge.yml deleted file mode 100644 index d4a00ae..0000000 --- a/.github/workflows/dependabot-auto-merge.yml +++ /dev/null @@ -1,18 +0,0 @@ -name: Dependabot Auto-merge - -on: pull_request - -permissions: - contents: write - pull-requests: write - -jobs: - auto-merge: - name: Enable auto-merge - runs-on: ubuntu-latest - if: github.actor == 'dependabot[bot]' - steps: - - name: Enable auto-merge - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: gh pr merge --auto --squash "${{ github.event.pull_request.number }}" --repo ${{ github.repository }} diff --git a/.github/workflows/labels.yml b/.github/workflows/labels.yml deleted file mode 100644 index 6455ba9..0000000 --- a/.github/workflows/labels.yml +++ /dev/null @@ -1,18 +0,0 @@ -name: Create Labels - -on: - workflow_dispatch: - -jobs: - labels: - name: Sync labels - runs-on: ubuntu-latest - permissions: - issues: write - steps: - - name: Create labels - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: | - gh label create dependencies --color 0075ca --description 'Dependency updates' --force --repo ${{ github.repository }} - gh label create ci --color e4e669 --description 'CI/CD changes' --force --repo ${{ github.repository }} diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml deleted file mode 100644 index 11fe113..0000000 --- a/.github/workflows/lint.yml +++ /dev/null @@ -1,31 +0,0 @@ -name: Lint & Type Check - -on: - pull_request: - branches: - - main - -jobs: - lint: - name: Lint & type check - runs-on: ubuntu-latest - - steps: - - name: Checkout - uses: actions/checkout@v6 - - - name: Set up Python - uses: actions/setup-python@v6 - with: - python-version: "3.12" - - - name: Install dependencies - run: | - pip install -e . - pip install flake8 mypy - - - name: flake8 - run: flake8 temporal_mcp/ - - - name: mypy - run: mypy temporal_mcp/ diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml deleted file mode 100644 index 148a4e6..0000000 --- a/.github/workflows/release.yml +++ /dev/null @@ -1,77 +0,0 @@ -name: Release - -on: - push: - branches: - - main - -jobs: - release: - name: Semantic release - runs-on: ubuntu-latest - concurrency: release - permissions: - contents: write - outputs: - released: ${{ steps.version_check.outputs.will_release }} - - steps: - - name: Checkout - uses: actions/checkout@v6 - with: - fetch-depth: 0 - token: ${{ secrets.GITHUB_TOKEN }} - - - name: Set up Python - uses: actions/setup-python@v6 - with: - python-version: "3.12" - - - name: Install tools - run: pip install python-semantic-release build - - - name: Check for releasable commits - id: version_check - run: | - version=$(semantic-release version --print 2>/dev/null || true) - if [ -n "$version" ]; then - echo "will_release=true" >> $GITHUB_OUTPUT - else - echo "will_release=false" >> $GITHUB_OUTPUT - fi - - - name: Run semantic release - if: steps.version_check.outputs.will_release == 'true' - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: semantic-release version --push - - - name: Build package - if: steps.version_check.outputs.will_release == 'true' - run: python -m build - - - name: Upload build artifacts - if: steps.version_check.outputs.will_release == 'true' - uses: actions/upload-artifact@v7 - with: - name: dist - path: dist/ - - publish: - name: Publish to PyPI - needs: release - if: needs.release.outputs.released == 'true' - runs-on: ubuntu-latest - environment: pypi - permissions: - id-token: write - - steps: - - name: Download build artifacts - uses: actions/download-artifact@v8 - with: - name: dist - path: dist/ - - - name: Publish to PyPI - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml deleted file mode 100644 index eea53a4..0000000 --- a/.github/workflows/security.yml +++ /dev/null @@ -1,43 +0,0 @@ -name: Security - -on: - pull_request: - branches: - - main - push: - branches: - - main - schedule: - - cron: "0 8 * * 1" # Every Monday at 08:00 UTC - -jobs: - pip-audit: - name: Dependency vulnerability scan - runs-on: ubuntu-latest - - steps: - - name: Checkout - uses: actions/checkout@v6 - - - name: Set up Python - uses: actions/setup-python@v6 - with: - python-version: "3.12" - - - name: Run pip-audit - run: | - pip install pip-audit - pip-audit -r requirements.txt - - semgrep: - name: SAST (Semgrep) - runs-on: ubuntu-latest - container: - image: semgrep/semgrep - - steps: - - name: Checkout - uses: actions/checkout@v6 - - - name: Run Semgrep - run: semgrep --config=auto --error temporal_mcp/ diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml deleted file mode 100644 index 726831b..0000000 --- a/.github/workflows/test.yml +++ /dev/null @@ -1,32 +0,0 @@ -name: Tests - -on: - pull_request: - branches: - - main - -jobs: - test: - name: Run test suite - runs-on: ubuntu-latest - - strategy: - matrix: - python-version: ["3.10", "3.11", "3.12", "3.13"] - - steps: - - name: Checkout - uses: actions/checkout@v6 - - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v6 - with: - python-version: ${{ matrix.python-version }} - - - name: Install dependencies - run: | - pip install -e . - pip install pytest pytest-asyncio - - - name: Run tests - run: pytest tests/ -v diff --git a/README.md b/README.md index ba6a856..67b222a 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,8 @@ -# Temporal MCP Server +# Temporal MCP Server — LoyaltyLion read-only fork + +> **This is a fork.** Upstream is [GethosTheWalrus/temporal-mcp](https://github.com/GethosTheWalrus/temporal-mcp). This fork removes every mutating tool (start / signal / cancel / terminate / continue_as_new, the `batch_*` trio, and the schedule mutations) so AI clients can only inspect Temporal — they can't change anything. Only six tools remain: `describe_workflow`, `get_workflow_history`, `get_workflow_result`, `list_schedules`, `list_workflows`, `query_workflow`. See `temporal_mcp/tools/tool_definitions.py` and `temporal_mcp/server.py` for the trimmed surface. +> +> The PyPI / Docker Hub distributions linked below are upstream and have the full tool set — do not use them. Install from this fork (`pip install git+https://github.com/loyaltylion/temporal-mcp@`). ## Overview @@ -12,36 +16,19 @@ Read more on the [Temporal Code Exchange](https://temporal.io/code-exchange/temp ## Tools -### Workflow Execution +This fork exposes six read-only tools. Every other tool from upstream has been removed at the source level. + +### Workflow Inspection -- **`start_workflow`** - Start a new Temporal workflow execution with specified parameters, workflow ID, and task queue -- **`get_workflow_result`** - Retrieve the result of a completed workflow execution - **`describe_workflow`** - Get detailed information about a workflow execution including status, timing, and metadata -- **`list_workflows`** - List workflow executions based on a query filter with pagination support (limit/skip) +- **`get_workflow_result`** - Retrieve the result of a completed workflow execution - **`get_workflow_history`** - Retrieve the complete event history of a workflow execution +- **`list_workflows`** - List workflow executions based on a query filter with pagination support (limit/skip) +- **`query_workflow`** - Query a running workflow for its current state. Read-only by Temporal contract — queries don't append history events or fire activities, though a buggy workflow-side query handler could mutate in-memory state. -### Workflow Control - -- **`query_workflow`** - Query a running workflow for its current state without affecting execution -- **`signal_workflow`** - Send a signal to a running workflow to change its behavior or provide data -- **`cancel_workflow`** - Request cancellation of a running workflow execution -- **`terminate_workflow`** - Forcefully terminate a workflow execution with a reason -- **`continue_as_new`** - Signal a workflow to continue as new (restart with new inputs while preserving history link) - -### Batch Operations - -- **`batch_signal`** - Send a signal to multiple workflows matching a query (configurable batch size) -- **`batch_cancel`** - Cancel multiple workflows matching a query (configurable batch size) -- **`batch_terminate`** - Terminate multiple workflows matching a query with a specified reason (configurable batch size) - -### Schedule Management +### Schedule Inspection -- **`create_schedule`** - Create a new schedule for periodic workflow execution using cron expressions - **`list_schedules`** - List all schedules with pagination support (limit/skip) -- **`pause_schedule`** - Pause a schedule to temporarily stop workflow executions -- **`unpause_schedule`** - Resume a paused schedule -- **`delete_schedule`** - Permanently delete a schedule -- **`trigger_schedule`** - Manually trigger a scheduled workflow immediately ## Temporal Documentation diff --git a/pyproject.toml b/pyproject.toml index e0a93c3..1418db7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,14 +1,14 @@ [project] name = "temporal-mcp-server" version = "1.1.1" -description = "MCP server for Temporal workflow orchestration" +description = "MCP server for Temporal workflow orchestration (LoyaltyLion read-only fork)" readme = "README.md" license = {text = "Apache-2.0"} requires-python = ">=3.10" authors = [ {name = "Mike Toscano", email = "mike@miketoscano.com"}, ] -keywords = ["temporal", "mcp", "model-context-protocol", "workflow", "orchestration"] +keywords = ["temporal", "mcp", "model-context-protocol", "workflow", "orchestration", "read-only"] classifiers = [ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", @@ -27,9 +27,9 @@ dependencies = [ ] [project.urls] -Homepage = "https://github.com/GethosTheWalrus/temporal-mcp" -Repository = "https://github.com/GethosTheWalrus/temporal-mcp" -Issues = "https://github.com/GethosTheWalrus/temporal-mcp/issues" +Homepage = "https://github.com/loyaltylion/temporal-mcp" +Repository = "https://github.com/loyaltylion/temporal-mcp" +Upstream = "https://github.com/GethosTheWalrus/temporal-mcp" [project.scripts] temporal-mcp-server = "temporal_mcp.__main__:main" diff --git a/temporal_mcp/handlers/batch_handlers.py b/temporal_mcp/handlers/batch_handlers.py deleted file mode 100644 index 923fde2..0000000 --- a/temporal_mcp/handlers/batch_handlers.py +++ /dev/null @@ -1,226 +0,0 @@ -"""Handlers for batch workflow operations.""" - -import json -import sys - -from mcp.types import TextContent -from temporalio.client import Client - - -async def batch_signal(client: Client, args: dict) -> list[TextContent]: - """Send signal to multiple workflows. - - Args: - client: Connected Temporal client - args: Arguments containing query, signal_name, optional args, and optional limit - - Returns: - Batch operation results with success and error counts - """ - query = args["query"] - signal_name = args["signal_name"] - signal_args = args.get("args") - limit = args.get("limit", 100) - - workflows_signaled: list[str] = [] - errors: list[dict[str, str]] = [] - - async for workflow in client.list_workflows(query): - if len(workflows_signaled) + len(errors) >= limit: - break - - try: - handle = client.get_workflow_handle(workflow.id) - await handle.signal(signal_name, signal_args) - workflows_signaled.append(workflow.id) - except Exception as e: - error_detail = {"workflow_id": workflow.id, "error": str(e), "error_type": type(e).__name__} - errors.append(error_detail) - print(f"Error signaling workflow {workflow.id}: {e}", file=sys.stderr) - - result = {"signal_name": signal_name, "workflows_signaled": workflows_signaled, "success_count": len(workflows_signaled), "error_count": len(errors)} - - if errors: - result["errors"] = errors - - return [TextContent(type="text", text=json.dumps(result, indent=2))] - - -async def batch_cancel(client: Client, args: dict) -> list[TextContent]: - """Cancel multiple workflows with concurrent processing. - - Args: - client: Connected Temporal client - args: Arguments containing query and optional limit, concurrency - - Returns: - Batch operation results with success and error counts - """ - import asyncio - - query = args["query"] - limit = args.get("limit", 100) - concurrency = args.get("concurrency", 50) # Process 50 workflows concurrently - - print(f"Starting batch cancel with limit={limit}, concurrency={concurrency}", file=sys.stderr) - - workflows_cancelled = [] - errors = [] - - async def cancel_workflow(workflow_id: str) -> tuple[str, Exception | None]: - """Cancel a single workflow and return result.""" - try: - handle = client.get_workflow_handle(workflow_id) - await handle.cancel() - return workflow_id, None - except Exception as e: - return workflow_id, e - - # Collect workflows to cancel - workflows_to_cancel = [] - async for workflow in client.list_workflows(query): - workflows_to_cancel.append(workflow.id) - if len(workflows_to_cancel) >= limit: - break - - total = len(workflows_to_cancel) - print(f"Found {total} workflows to cancel. Starting cancellation...", file=sys.stderr) - - # Process in batches for concurrency - for i in range(0, len(workflows_to_cancel), concurrency): - batch = workflows_to_cancel[i : i + concurrency] - batch_num = i // concurrency + 1 - total_batches = (len(workflows_to_cancel) + concurrency - 1) // concurrency - - print(f"Processing batch {batch_num}/{total_batches} ({len(batch)} workflows)...", file=sys.stderr) - - # Cancel workflows concurrently in this batch - results = await asyncio.gather(*[cancel_workflow(wf_id) for wf_id in batch], return_exceptions=False) - - # Process results - for workflow_id, error in results: - if error is None: - workflows_cancelled.append(workflow_id) - else: - error_detail = {"workflow_id": workflow_id, "error": str(error), "error_type": type(error).__name__} - errors.append(error_detail) - print(f"Error cancelling workflow {workflow_id}: {error}", file=sys.stderr) - - print(f"Batch {batch_num}/{total_batches} complete. Total cancelled: {len(workflows_cancelled)}, errors: {len(errors)}", file=sys.stderr) - - print(f"Batch cancel complete! Cancelled: {len(workflows_cancelled)}, Errors: {len(errors)}", file=sys.stderr) - - # Return only summary to avoid context overflow - do NOT include full list of IDs - result = { - "success_count": len(workflows_cancelled), - "error_count": len(errors), - "total_processed": len(workflows_cancelled) + len(errors), - "message": f"Successfully cancelled {len(workflows_cancelled)} workflows.", - } - - # Include first and last few IDs as samples only - if len(workflows_cancelled) > 0: - if len(workflows_cancelled) <= 10: - result["cancelled_workflows"] = workflows_cancelled - else: - result["sample_first"] = workflows_cancelled[:5] - result["sample_last"] = workflows_cancelled[-5:] - result["note"] = f"Showing first 5 and last 5 of {len(workflows_cancelled)} cancelled workflows to avoid context overflow" - - if errors: - result["sample_errors"] = errors[:5] # Only show first 5 errors - if len(errors) > 5: - result["errors_note"] = f"Showing first 5 of {len(errors)} errors" - - return [TextContent(type="text", text=json.dumps(result, indent=2))] - - -async def batch_terminate(client: Client, args: dict) -> list[TextContent]: - """Terminate multiple workflows with concurrent processing. - - Args: - client: Connected Temporal client - args: Arguments containing query, optional reason, limit, and concurrency - - Returns: - Batch operation results with success and error counts - """ - import asyncio - - query = args["query"] - reason = args.get("reason", "Batch termination via MCP") - limit = args.get("limit", 100) - concurrency = args.get("concurrency", 50) # Process 50 workflows concurrently - - print(f"Starting batch terminate with limit={limit}, concurrency={concurrency}", file=sys.stderr) - - workflows_terminated = [] - errors = [] - - async def terminate_workflow(workflow_id: str) -> tuple[str, Exception | None]: - """Terminate a single workflow and return result.""" - try: - handle = client.get_workflow_handle(workflow_id) - await handle.terminate(reason) - return workflow_id, None - except Exception as e: - return workflow_id, e - - # Collect workflows to terminate - workflows_to_terminate = [] - async for workflow in client.list_workflows(query): - workflows_to_terminate.append(workflow.id) - if len(workflows_to_terminate) >= limit: - break - - total = len(workflows_to_terminate) - print(f"Found {total} workflows to terminate. Starting termination...", file=sys.stderr) - - # Process in batches for concurrency - for i in range(0, len(workflows_to_terminate), concurrency): - batch = workflows_to_terminate[i : i + concurrency] - batch_num = i // concurrency + 1 - total_batches = (len(workflows_to_terminate) + concurrency - 1) // concurrency - - print(f"Processing batch {batch_num}/{total_batches} ({len(batch)} workflows)...", file=sys.stderr) - - # Terminate workflows concurrently in this batch - results = await asyncio.gather(*[terminate_workflow(wf_id) for wf_id in batch], return_exceptions=False) - - # Process results - for workflow_id, error in results: - if error is None: - workflows_terminated.append(workflow_id) - else: - error_detail = {"workflow_id": workflow_id, "error": str(error), "error_type": type(error).__name__} - errors.append(error_detail) - print(f"Error terminating workflow {workflow_id}: {error}", file=sys.stderr) - - print(f"Batch {batch_num}/{total_batches} complete. Total terminated: {len(workflows_terminated)}, errors: {len(errors)}", file=sys.stderr) - - print(f"Batch terminate complete! Terminated: {len(workflows_terminated)}, Errors: {len(errors)}", file=sys.stderr) - - # Return only summary to avoid context overflow - do NOT include full list of IDs - result = { - "reason": reason, - "success_count": len(workflows_terminated), - "error_count": len(errors), - "total_processed": len(workflows_terminated) + len(errors), - "message": f"Successfully terminated {len(workflows_terminated)} workflows.", - } - - # Include first and last few IDs as samples only - if len(workflows_terminated) > 0: - if len(workflows_terminated) <= 10: - result["terminated_workflows"] = workflows_terminated - else: - result["sample_first"] = workflows_terminated[:5] - result["sample_last"] = workflows_terminated[-5:] - result["note"] = f"Showing first 5 and last 5 of {len(workflows_terminated)} terminated workflows to avoid context overflow" - - if errors: - result["sample_errors"] = errors[:5] # Only show first 5 errors - if len(errors) > 5: - result["errors_note"] = f"Showing first 5 of {len(errors)} errors" - - return [TextContent(type="text", text=json.dumps(result, indent=2))] diff --git a/temporal_mcp/handlers/query_handlers.py b/temporal_mcp/handlers/query_handlers.py index 0972e27..fa067d4 100644 --- a/temporal_mcp/handlers/query_handlers.py +++ b/temporal_mcp/handlers/query_handlers.py @@ -1,4 +1,9 @@ -"""Handlers for workflow query and signal operations.""" +"""Handlers for workflow query operations. + +LoyaltyLion read-only fork: signal_workflow and continue_as_new have been +removed. Only query_workflow remains; queries are read-only by Temporal +contract. +""" import json @@ -24,51 +29,3 @@ async def query_workflow(client: Client, args: dict) -> list[TextContent]: result = await handle.query(query_name, query_args) return [TextContent(type="text", text=json.dumps({"query_result": result}, indent=2, default=str))] - - -async def signal_workflow(client: Client, args: dict) -> list[TextContent]: - """Send a signal to a workflow. - - Args: - client: Connected Temporal client - args: Arguments containing workflow_id, signal_name, and optional args - - Returns: - Success response - """ - workflow_id = args["workflow_id"] - signal_name = args["signal_name"] - signal_args = args.get("args") - - handle = client.get_workflow_handle(workflow_id) - await handle.signal(signal_name, signal_args) - - return [TextContent(type="text", text=json.dumps({"status": "signal_sent", "workflow_id": workflow_id, "signal_name": signal_name}, indent=2))] - - -async def continue_as_new(client: Client, args: dict) -> list[TextContent]: - """Signal a workflow to continue as new. - - Note: This sends a signal to the workflow. The workflow itself must be - designed to call workflow.continue_as_new() when it receives this signal. - - Args: - client: Connected Temporal client - args: Arguments containing workflow_id, signal_name, and optional signal_args - - Returns: - Success response - """ - workflow_id = args["workflow_id"] - signal_name = args["signal_name"] - signal_args = args.get("signal_args", {}) - - handle = client.get_workflow_handle(workflow_id) - await handle.signal(signal_name, signal_args) - - return [ - TextContent( - type="text", - text=json.dumps({"status": "signal_sent", "workflow_id": workflow_id, "signal_name": signal_name, "note": "Workflow must implement continue-as-new logic in signal handler"}, indent=2), - ) - ] diff --git a/temporal_mcp/handlers/schedule_handlers.py b/temporal_mcp/handlers/schedule_handlers.py index 6d03d83..6048f82 100644 --- a/temporal_mcp/handlers/schedule_handlers.py +++ b/temporal_mcp/handlers/schedule_handlers.py @@ -1,42 +1,14 @@ -"""Handlers for schedule operations.""" +"""Handlers for schedule operations. + +LoyaltyLion read-only fork: only list_schedules is exposed. The schedule +mutation handlers (create/pause/unpause/delete/trigger) have been removed. +""" import json import sys from mcp.types import TextContent -from temporalio.client import Client, Schedule, ScheduleActionStartWorkflow, ScheduleSpec - - -async def create_schedule(client: Client, args: dict) -> list[TextContent]: - """Create a new workflow schedule. - - Args: - client: Connected Temporal client - args: Arguments containing schedule_id, workflow_name, task_queue, cron, and optional args - - Returns: - Success response with schedule details - """ - schedule_id = args["schedule_id"] - workflow_name = args["workflow_name"] - task_queue = args["task_queue"] - cron = args["cron"] - workflow_args = args.get("args", {}) - - await client.create_schedule( - schedule_id, - Schedule( - action=ScheduleActionStartWorkflow( - workflow_name, - workflow_args, - id=f"{schedule_id}-workflow", - task_queue=task_queue, - ), - spec=ScheduleSpec(cron_expressions=[cron]), - ), - ) - - return [TextContent(type="text", text=json.dumps({"status": "created", "schedule_id": schedule_id, "workflow_name": workflow_name, "cron": cron}, indent=2))] +from temporalio.client import Client async def list_schedules(client: Client, args: dict) -> list[TextContent]: @@ -57,7 +29,6 @@ async def list_schedules(client: Client, args: dict) -> list[TextContent]: total_fetched = 0 async for schedule in await client.list_schedules(): - # Skip the first 'skip' results if count < skip: count += 1 continue @@ -74,7 +45,6 @@ async def list_schedules(client: Client, args: dict) -> list[TextContent]: if total_fetched >= limit: break - # Check if there are more results has_more = False try: async for _ in await client.list_schedules(): @@ -97,77 +67,3 @@ async def list_schedules(client: Client, args: dict) -> list[TextContent]: result["message"] = f"Showing all {len(schedules)} schedules (skipped {skip}). No more results." return [TextContent(type="text", text=json.dumps(result, indent=2))] - - -async def pause_schedule(client: Client, args: dict) -> list[TextContent]: - """Pause a schedule. - - Args: - client: Connected Temporal client - args: Arguments containing schedule_id and optional note - - Returns: - Success response - """ - schedule_id = args["schedule_id"] - note = args.get("note", "Paused via MCP") - - handle = client.get_schedule_handle(schedule_id) - await handle.pause(note=note) - - return [TextContent(type="text", text=json.dumps({"status": "paused", "schedule_id": schedule_id, "note": note}, indent=2))] - - -async def unpause_schedule(client: Client, args: dict) -> list[TextContent]: - """Resume a paused schedule. - - Args: - client: Connected Temporal client - args: Arguments containing schedule_id and optional note - - Returns: - Success response - """ - schedule_id = args["schedule_id"] - note = args.get("note", "Resumed via MCP") - - handle = client.get_schedule_handle(schedule_id) - await handle.unpause(note=note) - - return [TextContent(type="text", text=json.dumps({"status": "unpaused", "schedule_id": schedule_id, "note": note}, indent=2))] - - -async def delete_schedule(client: Client, args: dict) -> list[TextContent]: - """Delete a schedule. - - Args: - client: Connected Temporal client - args: Arguments containing schedule_id - - Returns: - Success response - """ - schedule_id = args["schedule_id"] - - handle = client.get_schedule_handle(schedule_id) - await handle.delete() - - return [TextContent(type="text", text=json.dumps({"status": "deleted", "schedule_id": schedule_id}, indent=2))] - - -async def trigger_schedule(client: Client, args: dict) -> list[TextContent]: - """Manually trigger a schedule. - - Args: - client: Connected Temporal client - args: Arguments containing schedule_id - - Returns: - Success response - """ - schedule_id = args["schedule_id"] - - handle = client.get_schedule_handle(schedule_id) - await handle.trigger() - - return [TextContent(type="text", text=json.dumps({"status": "triggered", "schedule_id": schedule_id}, indent=2))] diff --git a/temporal_mcp/handlers/workflow_handlers.py b/temporal_mcp/handlers/workflow_handlers.py index fb90dc2..1ad203f 100644 --- a/temporal_mcp/handlers/workflow_handlers.py +++ b/temporal_mcp/handlers/workflow_handlers.py @@ -1,4 +1,8 @@ -"""Handlers for workflow operations.""" +"""Handlers for workflow operations. + +LoyaltyLion read-only fork: write handlers (start_workflow, +cancel_workflow, terminate_workflow) have been removed. +""" import asyncio import json @@ -9,69 +13,6 @@ from temporalio.api.enums.v1 import WorkflowExecutionStatus -async def start_workflow(client: Client, args: dict) -> list[TextContent]: - """Start a new workflow execution. - - Args: - client: Connected Temporal client - args: Arguments containing workflow_name, workflow_id, task_queue, and optional args - - Returns: - Success response with workflow details - """ - workflow_name = args["workflow_name"] - workflow_id = args["workflow_id"] - task_queue = args["task_queue"] - workflow_args = args.get("args", {}) - - handle = await client.start_workflow( - workflow_name, - workflow_args, - id=workflow_id, - task_queue=task_queue, - ) - - result = {"workflow_id": handle.id, "run_id": handle.result_run_id, "status": "started"} - return [TextContent(type="text", text=json.dumps(result, indent=2))] - - -async def cancel_workflow(client: Client, args: dict) -> list[TextContent]: - """Cancel a workflow execution. - - Args: - client: Connected Temporal client - args: Arguments containing workflow_id - - Returns: - Success response - """ - workflow_id = args["workflow_id"] - - handle = client.get_workflow_handle(workflow_id) - await handle.cancel() - - return [TextContent(type="text", text=json.dumps({"status": "cancelled", "workflow_id": workflow_id}, indent=2))] - - -async def terminate_workflow(client: Client, args: dict) -> list[TextContent]: - """Terminate a workflow execution. - - Args: - client: Connected Temporal client - args: Arguments containing workflow_id and optional reason - - Returns: - Success response - """ - workflow_id = args["workflow_id"] - reason = args.get("reason", "Terminated via MCP") - - handle = client.get_workflow_handle(workflow_id) - await handle.terminate(reason) - - return [TextContent(type="text", text=json.dumps({"status": "terminated", "workflow_id": workflow_id, "reason": reason}, indent=2))] - - async def get_workflow_result(client: Client, args: dict) -> list[TextContent]: """Get the result of a workflow execution. @@ -161,7 +102,6 @@ async def list_workflows(client: Client, args: dict) -> list[TextContent]: total_fetched = 0 async for workflow in client.list_workflows(query): - # Skip the first 'skip' results if count < skip: count += 1 continue @@ -182,7 +122,6 @@ async def list_workflows(client: Client, args: dict) -> list[TextContent]: if total_fetched >= limit: break - # Check if there are more results has_more = False try: async for _ in client.list_workflows(query): diff --git a/temporal_mcp/server.py b/temporal_mcp/server.py index 2464894..be1a7b3 100644 --- a/temporal_mcp/server.py +++ b/temporal_mcp/server.py @@ -1,4 +1,9 @@ -"""Main MCP Server for Temporal workflow orchestration.""" +"""Main MCP Server for Temporal workflow orchestration. + +LoyaltyLion read-only fork: only inspection tools are dispatched. Write +tools have been removed from the source — this dispatcher only references +the read handlers. +""" import json from typing import Any, Optional @@ -11,10 +16,8 @@ from .tools.tool_definitions import get_all_tools from .utils.exceptions import format_connection_error, format_error_response -# Import all handlers from .handlers import workflow_handlers from .handlers import query_handlers -from .handlers import batch_handlers from .handlers import schedule_handlers @@ -62,61 +65,26 @@ async def list_tools() -> list[Tool]: @self.server.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """Handle tool execution requests.""" - # Ensure connection try: await self.client_manager.connect() except Exception as e: return format_connection_error(e) - # Route to appropriate handler try: client = self.client_manager.ensure_connected() - # Workflow operations - if name == "start_workflow": - return await workflow_handlers.start_workflow(client, arguments) - elif name == "cancel_workflow": - return await workflow_handlers.cancel_workflow(client, arguments) - elif name == "terminate_workflow": - return await workflow_handlers.terminate_workflow(client, arguments) + if name == "describe_workflow": + return await workflow_handlers.describe_workflow(client, arguments) + elif name == "get_workflow_history": + return await workflow_handlers.get_workflow_history(client, arguments) elif name == "get_workflow_result": return await workflow_handlers.get_workflow_result(client, arguments) - elif name == "describe_workflow": - return await workflow_handlers.describe_workflow(client, arguments) elif name == "list_workflows": return await workflow_handlers.list_workflows(client, arguments) - elif name == "get_workflow_history": - return await workflow_handlers.get_workflow_history(client, arguments) - - # Query and signal operations elif name == "query_workflow": return await query_handlers.query_workflow(client, arguments) - elif name == "signal_workflow": - return await query_handlers.signal_workflow(client, arguments) - elif name == "continue_as_new": - return await query_handlers.continue_as_new(client, arguments) - - # Batch operations - elif name == "batch_signal": - return await batch_handlers.batch_signal(client, arguments) - elif name == "batch_cancel": - return await batch_handlers.batch_cancel(client, arguments) - elif name == "batch_terminate": - return await batch_handlers.batch_terminate(client, arguments) - - # Schedule operations - elif name == "create_schedule": - return await schedule_handlers.create_schedule(client, arguments) elif name == "list_schedules": return await schedule_handlers.list_schedules(client, arguments) - elif name == "pause_schedule": - return await schedule_handlers.pause_schedule(client, arguments) - elif name == "unpause_schedule": - return await schedule_handlers.unpause_schedule(client, arguments) - elif name == "delete_schedule": - return await schedule_handlers.delete_schedule(client, arguments) - elif name == "trigger_schedule": - return await schedule_handlers.trigger_schedule(client, arguments) else: return [TextContent(type="text", text=json.dumps({"error": f"Unknown tool: {name}", "type": "unknown_tool"}, indent=2))] diff --git a/temporal_mcp/tools/tool_definitions.py b/temporal_mcp/tools/tool_definitions.py index ae51235..d4be14e 100644 --- a/temporal_mcp/tools/tool_definitions.py +++ b/temporal_mcp/tools/tool_definitions.py @@ -1,4 +1,10 @@ -"""Tool definitions for the Temporal MCP server.""" +"""Tool definitions for the Temporal MCP server. + +LoyaltyLion read-only fork: only inspection tools are exposed. +Write tools (start/signal/cancel/terminate/continue_as_new, the batch_* +trio, and the schedule mutations) have been removed from this fork. See +fork README for rationale. +""" from mcp.types import Tool @@ -10,82 +16,11 @@ def get_all_tools() -> list[Tool]: List of Tool definitions """ return [ - Tool( - name="start_workflow", - description="Start a new Temporal workflow execution", - inputSchema={ - "type": "object", - "properties": { - "workflow_name": {"type": "string", "description": "The name of the workflow to start"}, - "workflow_id": {"type": "string", "description": "Unique identifier for the workflow execution"}, - "task_queue": {"type": "string", "description": "The task queue to use for this workflow"}, - "args": {"type": "object", "description": "Arguments to pass to the workflow (as JSON object)"}, - }, - "required": ["workflow_name", "workflow_id", "task_queue"], - }, - ), - Tool( - name="query_workflow", - description="Query a running workflow for its current state", - inputSchema={ - "type": "object", - "properties": { - "workflow_id": {"type": "string", "description": "The workflow execution ID to query"}, - "query_name": {"type": "string", "description": "The name of the query to execute"}, - "args": {"type": "object", "description": "Arguments for the query (as JSON object)"}, - }, - "required": ["workflow_id", "query_name"], - }, - ), - Tool( - name="signal_workflow", - description="Send a signal to a running workflow", - inputSchema={ - "type": "object", - "properties": { - "workflow_id": {"type": "string", "description": "The workflow execution ID to signal"}, - "signal_name": {"type": "string", "description": "The name of the signal to send"}, - "args": {"type": "object", "description": "Arguments for the signal (as JSON object)"}, - }, - "required": ["workflow_id", "signal_name"], - }, - ), - Tool( - name="cancel_workflow", - description="Cancel a running workflow execution", - inputSchema={"type": "object", "properties": {"workflow_id": {"type": "string", "description": "The workflow execution ID to cancel"}}, "required": ["workflow_id"]}, - ), - Tool( - name="get_workflow_result", - description="Get the result of a completed workflow", - inputSchema={"type": "object", "properties": {"workflow_id": {"type": "string", "description": "The workflow execution ID"}}, "required": ["workflow_id"]}, - ), Tool( name="describe_workflow", description="Get detailed information about a workflow execution", inputSchema={"type": "object", "properties": {"workflow_id": {"type": "string", "description": "The workflow execution ID to describe"}}, "required": ["workflow_id"]}, ), - Tool( - name="list_workflows", - description="List workflow executions based on a query. Specify 'limit' to control the number of results (default: 100, max recommended: 1000). Use 'skip' to paginate through results.", - inputSchema={ - "type": "object", - "properties": { - "query": {"type": "string", "description": "List filter query (e.g., 'WorkflowType=\"MyWorkflow\"')"}, - "limit": {"type": "number", "description": "Maximum number of results to return (default: 100, increase for more results)"}, - "skip": {"type": "number", "description": "Number of results to skip for pagination (default: 0)"}, - }, - }, - ), - Tool( - name="terminate_workflow", - description="Forcefully terminate a workflow execution", - inputSchema={ - "type": "object", - "properties": {"workflow_id": {"type": "string", "description": "The workflow execution ID to terminate"}, "reason": {"type": "string", "description": "Reason for termination"}}, - "required": ["workflow_id"], - }, - ), Tool( name="get_workflow_history", description="Get the complete event history of a workflow execution. Specify 'limit' to control the number of events (default: 1000).", @@ -99,59 +34,9 @@ def get_all_tools() -> list[Tool]: }, ), Tool( - name="batch_signal", - description="Send a signal to multiple workflows matching a query. Specify 'limit' to control batch size (default: 100).", - inputSchema={ - "type": "object", - "properties": { - "query": {"type": "string", "description": "Query to select workflows"}, - "signal_name": {"type": "string", "description": "The signal name to send"}, - "args": {"type": "object", "description": "Arguments for the signal"}, - "limit": {"type": "number", "description": "Maximum number of workflows to signal (default: 100)"}, - }, - "required": ["query", "signal_name"], - }, - ), - Tool( - name="batch_cancel", - description="Cancel multiple workflows matching a query with concurrent processing for speed. Use 'concurrency' to control parallel operations (default: 50).", - inputSchema={ - "type": "object", - "properties": { - "query": {"type": "string", "description": "Query to select workflows to cancel"}, - "limit": {"type": "number", "description": "Maximum number of workflows to cancel (default: 100)"}, - "concurrency": {"type": "number", "description": "Number of workflows to cancel concurrently for faster processing (default: 50, max recommended: 100)"}, - }, - "required": ["query"], - }, - ), - Tool( - name="batch_terminate", - description="Terminate multiple workflows matching a query. Specify 'limit' to control batch size (default: 100).", - inputSchema={ - "type": "object", - "properties": { - "query": {"type": "string", "description": "Query to select workflows to terminate"}, - "reason": {"type": "string", "description": "Reason for termination"}, - "limit": {"type": "number", "description": "Maximum number of workflows to terminate (default: 100)"}, - }, - "required": ["query"], - }, - ), - Tool( - name="create_schedule", - description="Create a new schedule for periodic workflow execution", - inputSchema={ - "type": "object", - "properties": { - "schedule_id": {"type": "string", "description": "Unique identifier for the schedule"}, - "workflow_name": {"type": "string", "description": "Name of the workflow to schedule"}, - "task_queue": {"type": "string", "description": "Task queue for the workflow"}, - "cron": {"type": "string", "description": "Cron expression (e.g., '0 12 * * *')"}, - "args": {"type": "object", "description": "Arguments for the workflow"}, - }, - "required": ["schedule_id", "workflow_name", "task_queue", "cron"], - }, + name="get_workflow_result", + description="Get the result of a completed workflow", + inputSchema={"type": "object", "properties": {"workflow_id": {"type": "string", "description": "The workflow execution ID"}}, "required": ["workflow_id"]}, ), Tool( name="list_schedules", @@ -165,47 +50,28 @@ def get_all_tools() -> list[Tool]: }, ), Tool( - name="pause_schedule", - description="Pause a schedule", - inputSchema={ - "type": "object", - "properties": {"schedule_id": {"type": "string", "description": "The schedule ID to pause"}, "note": {"type": "string", "description": "Note explaining why the schedule was paused"}}, - "required": ["schedule_id"], - }, - ), - Tool( - name="unpause_schedule", - description="Resume a paused schedule", + name="list_workflows", + description="List workflow executions based on a query. Specify 'limit' to control the number of results (default: 100, max recommended: 1000). Use 'skip' to paginate through results.", inputSchema={ "type": "object", "properties": { - "schedule_id": {"type": "string", "description": "The schedule ID to unpause"}, - "note": {"type": "string", "description": "Note explaining why the schedule was resumed"}, + "query": {"type": "string", "description": "List filter query (e.g., 'WorkflowType=\"MyWorkflow\"')"}, + "limit": {"type": "number", "description": "Maximum number of results to return (default: 100, increase for more results)"}, + "skip": {"type": "number", "description": "Number of results to skip for pagination (default: 0)"}, }, - "required": ["schedule_id"], }, ), Tool( - name="delete_schedule", - description="Delete a schedule", - inputSchema={"type": "object", "properties": {"schedule_id": {"type": "string", "description": "The schedule ID to delete"}}, "required": ["schedule_id"]}, - ), - Tool( - name="trigger_schedule", - description="Manually trigger a scheduled workflow immediately", - inputSchema={"type": "object", "properties": {"schedule_id": {"type": "string", "description": "The schedule ID to trigger"}}, "required": ["schedule_id"]}, - ), - Tool( - name="continue_as_new", - description="Signal a workflow to continue as new (restart with new inputs while preserving history link)", + name="query_workflow", + description="Query a running workflow for its current state", inputSchema={ "type": "object", "properties": { - "workflow_id": {"type": "string", "description": "The workflow ID to continue as new"}, - "signal_name": {"type": "string", "description": "The signal name to send (must be handled by the workflow to trigger continue-as-new)"}, - "signal_args": {"type": "object", "description": "Arguments for the signal that will trigger continue-as-new"}, + "workflow_id": {"type": "string", "description": "The workflow execution ID to query"}, + "query_name": {"type": "string", "description": "The name of the query to execute"}, + "args": {"type": "object", "description": "Arguments for the query (as JSON object)"}, }, - "required": ["workflow_id", "signal_name"], + "required": ["workflow_id", "query_name"], }, ), ] diff --git a/tests/test_batch_handlers.py b/tests/test_batch_handlers.py deleted file mode 100644 index a85b587..0000000 --- a/tests/test_batch_handlers.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Tests for batch handler tools.""" - -import json -import pytest -from unittest.mock import AsyncMock, MagicMock - -from temporal_mcp.handlers import batch_handlers - - -class TestBatchSignal: - @pytest.mark.asyncio - async def test_batch_signal_success(self, mock_client): - mock_wf1 = MagicMock() - mock_wf1.id = "workflow-1" - mock_wf2 = MagicMock() - mock_wf2.id = "workflow-2" - - async def mock_list_workflows(query): - for wf in [mock_wf1, mock_wf2]: - yield wf - - mock_client.list_workflows = mock_list_workflows - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = { - "query": "WorkflowType='TestWorkflow'", - "signal_name": "pause", - "args": {}, - "limit": 10, - } - - result = await batch_handlers.batch_signal(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["signal_name"] == "pause" - assert len(response["workflows_signaled"]) == 2 - assert "workflow-1" in response["workflows_signaled"] - assert "workflow-2" in response["workflows_signaled"] - - -class TestBatchCancel: - @pytest.mark.asyncio - async def test_batch_cancel_success(self, mock_client): - mock_wf1 = MagicMock() - mock_wf1.id = "workflow-1" - - async def mock_list_workflows(query): - yield mock_wf1 - - mock_client.list_workflows = mock_list_workflows - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = {"query": "WorkflowType='TestWorkflow'", "limit": 10} - - result = await batch_handlers.batch_cancel(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert len(response["cancelled_workflows"]) == 1 - assert "workflow-1" in response["cancelled_workflows"] - - -class TestBatchTerminate: - @pytest.mark.asyncio - async def test_batch_terminate_success(self, mock_client): - mock_wf1 = MagicMock() - mock_wf1.id = "workflow-1" - - async def mock_list_workflows(query): - yield mock_wf1 - - mock_client.list_workflows = mock_list_workflows - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = { - "query": "WorkflowType='TestWorkflow'", - "reason": "Batch cleanup", - "limit": 10, - } - - result = await batch_handlers.batch_terminate(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert len(response["terminated_workflows"]) == 1 - assert response["reason"] == "Batch cleanup" diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index cfa54a0..73d1847 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -7,19 +7,6 @@ class TestErrorHandling: - @pytest.mark.asyncio - async def test_start_workflow_error(self, mock_client): - mock_client.start_workflow.side_effect = Exception("Connection error") - - args = { - "workflow_name": "TestWorkflow", - "workflow_id": "test-workflow-123", - "task_queue": "test-queue", - } - - with pytest.raises(Exception, match="Connection error"): - await workflow_handlers.start_workflow(mock_client, args) - @pytest.mark.asyncio async def test_query_workflow_not_found(self, mock_client): mock_handle = AsyncMock() diff --git a/tests/test_query_handlers.py b/tests/test_query_handlers.py index 5b74ae8..1378078 100644 --- a/tests/test_query_handlers.py +++ b/tests/test_query_handlers.py @@ -26,45 +26,3 @@ async def test_query_workflow_success(self, mock_client): response = json.loads(result[0].text) assert response["query_result"]["status"] == "running" assert response["query_result"]["progress"] == 50 - - -class TestSignalWorkflow: - @pytest.mark.asyncio - async def test_signal_workflow_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = { - "workflow_id": "test-workflow-123", - "signal_name": "update_status", - "args": {"new_status": "paused"}, - } - - result = await query_handlers.signal_workflow(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "signal_sent" - mock_handle.signal.assert_called_once_with("update_status", {"new_status": "paused"}) - - -class TestContinueAsNew: - @pytest.mark.asyncio - async def test_continue_as_new_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = { - "workflow_id": "test-workflow-123", - "signal_name": "continue_with_new_data", - "signal_args": {"new_param": "value"}, - } - - result = await query_handlers.continue_as_new(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "signal_sent" - assert response["workflow_id"] == "test-workflow-123" - assert "continue-as-new" in response["note"] - mock_handle.signal.assert_called_once_with("continue_with_new_data", {"new_param": "value"}) diff --git a/tests/test_schedule_handlers.py b/tests/test_schedule_handlers.py deleted file mode 100644 index 7cebfb6..0000000 --- a/tests/test_schedule_handlers.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Tests for schedule handler tools.""" - -import json -import pytest -from unittest.mock import AsyncMock, MagicMock - -from temporal_mcp.handlers import schedule_handlers - - -class TestScheduleHandlers: - @pytest.mark.asyncio - async def test_create_schedule_success(self, mock_client): - mock_client.create_schedule = AsyncMock() - - args = { - "schedule_id": "test-schedule", - "workflow_name": "TestWorkflow", - "task_queue": "test-queue", - "cron": "0 9 * * *", - "args": {}, - } - - result = await schedule_handlers.create_schedule(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "created" - assert response["schedule_id"] == "test-schedule" - assert response["cron"] == "0 9 * * *" - - @pytest.mark.asyncio - async def test_list_schedules_success(self, mock_client): - mock_schedule = MagicMock() - mock_schedule.id = "test-schedule" - mock_schedule.schedule.state.paused = False - - async def mock_list_schedules_inner(): - yield mock_schedule - - async def mock_list_schedules(): - return mock_list_schedules_inner() - - mock_client.list_schedules = mock_list_schedules - - result = await schedule_handlers.list_schedules(mock_client, {"limit": 20}) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert len(response["schedules"]) == 1 - assert response["schedules"][0]["schedule_id"] == "test-schedule" - assert response["schedules"][0]["paused"] is False - - @pytest.mark.asyncio - async def test_pause_schedule_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_schedule_handle = MagicMock(return_value=mock_handle) - - args = {"schedule_id": "test-schedule", "note": "Paused for maintenance"} - result = await schedule_handlers.pause_schedule(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "paused" - assert response["schedule_id"] == "test-schedule" - mock_handle.pause.assert_called_once_with(note="Paused for maintenance") - - @pytest.mark.asyncio - async def test_unpause_schedule_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_schedule_handle = MagicMock(return_value=mock_handle) - - args = {"schedule_id": "test-schedule", "note": "Maintenance complete"} - result = await schedule_handlers.unpause_schedule(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "unpaused" - mock_handle.unpause.assert_called_once_with(note="Maintenance complete") - - @pytest.mark.asyncio - async def test_delete_schedule_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_schedule_handle = MagicMock(return_value=mock_handle) - - result = await schedule_handlers.delete_schedule(mock_client, {"schedule_id": "test-schedule"}) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "deleted" - mock_handle.delete.assert_called_once() - - @pytest.mark.asyncio - async def test_trigger_schedule_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_schedule_handle = MagicMock(return_value=mock_handle) - - result = await schedule_handlers.trigger_schedule(mock_client, {"schedule_id": "test-schedule"}) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "triggered" - mock_handle.trigger.assert_called_once() diff --git a/tests/test_workflow_handlers.py b/tests/test_workflow_handlers.py index 16d5186..f5eeaab 100644 --- a/tests/test_workflow_handlers.py +++ b/tests/test_workflow_handlers.py @@ -8,61 +8,6 @@ from temporal_mcp.handlers import workflow_handlers -class TestStartWorkflow: - @pytest.mark.asyncio - async def test_start_workflow_success(self, mock_client): - mock_handle = AsyncMock() - mock_handle.id = "test-workflow-123" - mock_handle.result_run_id = "run-456" - mock_client.start_workflow.return_value = mock_handle - - args = { - "workflow_name": "TestWorkflow", - "workflow_id": "test-workflow-123", - "task_queue": "test-queue", - "args": {"key": "value"}, - } - - result = await workflow_handlers.start_workflow(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["workflow_id"] == "test-workflow-123" - assert response["run_id"] == "run-456" - assert response["status"] == "started" - mock_client.start_workflow.assert_called_once() - - -class TestCancelWorkflow: - @pytest.mark.asyncio - async def test_cancel_workflow_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - result = await workflow_handlers.cancel_workflow(mock_client, {"workflow_id": "test-workflow-123"}) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "cancelled" - mock_handle.cancel.assert_called_once() - - -class TestTerminateWorkflow: - @pytest.mark.asyncio - async def test_terminate_workflow_success(self, mock_client): - mock_handle = AsyncMock() - mock_client.get_workflow_handle = MagicMock(return_value=mock_handle) - - args = {"workflow_id": "test-workflow-123", "reason": "Test termination"} - result = await workflow_handlers.terminate_workflow(mock_client, args) - - assert len(result) == 1 - response = json.loads(result[0].text) - assert response["status"] == "terminated" - assert response["reason"] == "Test termination" - mock_handle.terminate.assert_called_once_with("Test termination") - - class TestGetWorkflowResult: @pytest.mark.asyncio async def test_get_workflow_result_success(self, mock_client):