Metadata-Version: 2.4
Name: mcp-celery
Version: 0.1.1
Summary: Expose Celery tasks as async MCP tools with automatic lifecycle management
Project-URL: Homepage, https://github.com/saleemasekrea000/mcp-celery
Project-URL: Repository, https://github.com/saleemasekrea000/mcp-celery
Project-URL: Issues, https://github.com/saleemasekrea000/mcp-celery/issues
Project-URL: Changelog, https://github.com/saleemasekrea000/mcp-celery/blob/main/CHANGELOG.md
Author-email: Saleem <saleem@ancileo.com>
License: MIT License
        
        Copyright (c) 2026 Saleem
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: async,celery,llm,long-running,mcp,model-context-protocol,tools
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: Celery
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: celery>=5.3.0
Requires-Dist: mcp>=1.0.0
Provides-Extra: dev
Requires-Dist: build>=1.2.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: redis>=5.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Requires-Dist: twine>=5.0.0; extra == 'dev'
Requires-Dist: ty>=0.0.29; extra == 'dev'
Description-Content-Type: text/markdown

# mcp-celery

**Expose Celery tasks as async MCP tools — without writing the boilerplate.**

Long-running work (reports, file processing, LLM pipelines) usually runs in Celery. Exposing those tasks to an MCP client means writing four tools per task by hand: one to start, one to poll status, one to fetch the result, one to cancel. `mcp-celery` does that generation for you. You register a Celery task once; four MCP tools (or a single MCP-native async operation) appear.

This README is written for a first-time reader. If you've never touched Celery or MCP before, start at [What this package does](#what-this-package-does) and follow it top to bottom.

---

## Table of contents

- [What this package does](#what-this-package-does)
- [Core concepts in one page](#core-concepts-in-one-page)
- [Quickstart — one command](#quickstart--one-command)
- [The three examples](#the-three-examples)
- [How it works internally](#how-it-works-internally)
- [Public API](#public-api)
- [Project layout](#project-layout)
- [Running the unit tests](#running-the-unit-tests)

---

## What this package does

### The problem

Without `mcp-celery`, exposing one long-running Celery task to an MCP client takes roughly 100 lines of boilerplate — one MCP tool for each lifecycle step:

```python
@mcp.tool(name="generate_report_start")
async def generate_report_start(user_id: str) -> str:
    result = generate_report.apply_async(kwargs={"user_id": user_id})
    return json.dumps({"task_id": result.id, "status": "PENDING"})

@mcp.tool(name="generate_report_status")
async def generate_report_status(task_id: str) -> str:
    result = AsyncResult(task_id, app=celery_app)
    status = CELERY_TO_STATUS.get(result.state, "PENDING")
    # ...state mapping, progress extraction, error handling...

@mcp.tool(name="generate_report_result")
async def generate_report_result(task_id: str) -> str: ...

@mcp.tool(name="generate_report_cancel")
async def generate_report_cancel(task_id: str) -> str: ...
```

Multiply that by every task and the surface area quickly becomes unmaintainable. The full baseline implementation is in [examples/baseline/server.py](examples/baseline/server.py).

### The solution

```python
from mcp_celery import AsyncToolServer

server = AsyncToolServer("my-server", celery_app=celery_app)
server.register_async_tool(generate_report, description="Generate a report")
```

That single `register_async_tool` call produces four MCP tools:

| Tool name                  | Purpose                                            |
|----------------------------|----------------------------------------------------|
| `generate_report_start`    | Dispatch the Celery task, return a **token**.      |
| `generate_report_status`   | Poll the task with the token.                      |
| `generate_report_result`   | Fetch the final result once the task is complete.  |
| `generate_report_cancel`   | Revoke a running task.                             |

A `token` is simply the Celery task id — the single handle a client carries across calls.

---

## Core concepts in one page

| Concept | One-line explanation |
|---|---|
| **Celery task** | A regular Python function that has been decorated with `@celery_app.task`. It runs in a worker process, not in your MCP server. |
| **MCP tool** | A callable function that an LLM / MCP client can invoke through the Model Context Protocol. |
| **Async tool (mcp-celery)** | A Celery task that `mcp-celery` has auto-wrapped into a group of MCP tools (polling) or a single MCP task operation (Approach 3). |
| **Token** | A string identifier — in practice, the Celery task id. Returned by `_start`, passed into every follow-up call. |
| **Backend (`AbstractBackend`)** | The thing that actually runs a task. The only concrete implementation today is `CeleryBackend`, but the interface is narrow so you could swap in Dramatiq, ARQ, or a stub. |
| **Exposure strategy (`AbstractExposureStrategy`)** | How a task is surfaced over MCP. `PollingExposureStrategy` generates the four tools; `OperationResourceStrategy` uses MCP's experimental Task + Resource API. |
| **`AsyncToolServer`** | Public entry point. Owns a `FastMCP` instance, a backend, a strategy, and a registry. |
| **`TaskStatus`** | Canonical lifecycle enum (`PENDING`, `RUNNING`, `COMPLETED`, `FAILED`, `CANCELLED`, `RETRYING`). All backends map their native states into these values via [lifecycle.py](src/mcp_celery/lifecycle.py). |

The layering is strict: `server → exposure/backend → schema`. The exposure strategy never imports a concrete backend, and the backend never imports a strategy.

---

## Quickstart — one command

Everything in `examples/` can be run through a single script: `examples/run.sh`. It handles Redis, the virtualenv, and the Celery worker for you.

```bash
# From the repo root:
bash examples/run.sh                  # defaults to 'with_package'
bash examples/run.sh baseline         # manual implementation (no mcp-celery)
bash examples/run.sh with_package     # mcp-celery with polling (Approach 1)
bash examples/run.sh approach3        # MCP-native tasks + resource URI
bash examples/run.sh approach3_notify # Approach 3 with push notifications
```

What the script does on your behalf:

1. **Redis.** If `localhost:6379` isn't accepting connections, it starts a Docker container named `redis-test`. On subsequent runs it reuses the same container.
2. **Virtualenv.** On the first run it invokes [`examples/setup.sh`](examples/setup.sh), which creates `examples/.venv/` and installs `mcp-celery` in editable mode.
3. **Worker.** It launches `celery -A tasks worker` for the chosen example in the background and waits until Celery prints `ready`.
4. **Client.** It runs the example's `client.py`.
5. **Cleanup.** On exit (or Ctrl+C) it stops the worker. The Redis container is kept running for speed; remove it with `docker rm -f redis-test` when you're done.

Requirements: `python3`, `bash`, and (first-time only, if Redis isn't already running) `docker`.

---

## The three examples

Each example under `examples/` runs the *same* Celery task (`generate_report`, which sleeps for 6 seconds and reports progress at 25 % / 60 % / 90 %). Only the MCP wiring changes.

### 1. `baseline/` — without the package

Hand-written MCP tools. About 100 lines for a single task. This is the "before" picture — everything `mcp-celery` is designed to eliminate.

### 2. `with_package/` — polling strategy (Approach 1)

Uses the default [`PollingExposureStrategy`](src/mcp_celery/exposure/polling.py). Four tools per task, token-based polling. The client calls `_start`, polls `_status`, then calls `_result`.

```
LLM → generate_report_start(user_id="u_1")     ← {"token": "abc", "status": "PENDING"}
LLM → generate_report_status(token="abc")      ← {"token": "abc", "status": "RUNNING", "progress": 0.6}
LLM → generate_report_result(token="abc")      ← {"token": "abc", "status": "COMPLETED", "result": {...}}
```

### 3. `approach3/` — MCP task operations + resource URI

Uses [`OperationResourceStrategy`](src/mcp_celery/exposure/operation_resource.py). Instead of four tools, each task becomes:

- **One MCP tool** that returns a `CreateTaskResult` (an MCP `Task` object plus a `resourceUri` in `_meta`).
- **Standard MCP task endpoints** — `tasks/get`, `tasks/cancel` — backed by a [`CeleryTaskStore`](src/mcp_celery/stores/celery_task_store.py) that lazily syncs MCP Task state with Celery state on every read.
- **A resource template** `mcp://results/<tool>/<task_id>` that returns the final result via `resources/read`.

The result is no longer conflated with the task lifecycle — progress lives on the task, payload lives in a resource.

The `approach3_notify` variant adds `notifications=True`. The server spawns a watcher coroutine per task that pushes `TaskStatusNotification` to the client whenever the status changes, so the client can `await` instead of polling. See [`server_notifications.py`](examples/approach3/server_notifications.py) and [`client_notifications.py`](examples/approach3/client_notifications.py).

### Expected output (Approach 1)

```
Registered tools: ['generate_report_start', 'generate_report_status',
                   'generate_report_result', 'generate_report_cancel']

--- Calling generate_report_start ---
Response: { "token": "b16a...", "status": "PENDING" }

--- Polling generate_report_status ---
  Poll 1: RUNNING  (60%)
  Poll 2: RUNNING  (90%)
  Poll 3: COMPLETED

--- Calling generate_report_result ---
Response: {
  "token": "b16a...",
  "status": "COMPLETED",
  "result": { "user_id": "u_demo_42", "revenue": 42000, "items": 17 }
}
```

The client spawns the MCP server as a subprocess over stdio — you only run `run.sh`.

---

## How it works internally

Reading order: start at [`server.py`](src/mcp_celery/server.py), then follow the data flow below.

### Registration

```text
server.register_async_tool(celery_task)          # or @server.async_tool
  ├─ _schema_from_signature(celery_task.run)     # build JSON schema from Python type hints
  ├─ AsyncToolDef(name, description, schema, task_ref, …)
  ├─ ToolRegistry.register(tool_def)             # keep it around so we can list registrations
  ├─ strategy.generate_tools(tool_def, backend)  # → list[GeneratedTool]
  ├─ for tool in generated: mcp.tool(...)(tool.handler)
  └─ strategy.setup(mcp, backend)                # optional: resources, handler overrides, etc.
```

The strategy owns all decisions about *what* MCP surface to expose. Whether it's four tools or one tool + resources + notifications is entirely a strategy concern — neither the server nor the backend knows.

### Polling strategy (Approach 1) — runtime

```text
LLM → <name>_start(**kwargs)
  PollingExposureStrategy._make_start → _build_start_handler
  → backend.dispatch(task_ref, kwargs)             # CeleryBackend.apply_async
  → format_response(token, TaskStatus.PENDING)

LLM → <name>_status(token)
  → backend.get_status(token)                      # AsyncResult + map_celery_state
  → format_response(token, status, progress, metadata)

LLM → <name>_result(token)
  → backend.get_result(token)
  → if COMPLETED: format_response(..., result=...)
  → if FAILED:    format_response(..., error=..., is_error=True)
  → otherwise:    format_response(..., is_error=True, "not yet complete")

LLM → <name>_cancel(token)
  → backend.cancel(token)                          # result.revoke(terminate=True)
  → format_response(token, TaskStatus.CANCELLED)
```

`format_response` ([schema.py](src/mcp_celery/schema.py)) is the single source of truth for response shape. FastMCP serializes the returned dict into `TextContent` automatically — you never hand-build `TextContent`.

### Operation + Resource strategy (Approach 3) — runtime

Setup installs three handler overrides on the low-level MCP server:

- `CallToolRequest` — intercepted; for our tools it dispatches to Celery, creates an MCP `Task` in the `CeleryTaskStore`, and returns a `CreateTaskResult` with `_meta.resourceUri` pointing at the result resource.
- `CancelTaskRequest` — cancels the MCP task (so the store state becomes `cancelled`) *before* revoking the Celery task (to avoid a race where the next `get_task` would see a terminal state and reject the cancellation).
- `GetTaskPayloadRequest` — returns a pointer `{"resourceUri": "mcp://results/<tool>/<task_id>"}` instead of the payload itself; the client is expected to read the resource.

Plus a resource template per tool whose reader calls `backend.get_result(task_id)` and returns JSON.

`CeleryTaskStore` is an `InMemoryTaskStore` that overrides `get_task` to query Celery on every read and reconcile the MCP task status with what the worker actually reports. Terminal MCP states are never overwritten.

When `notifications=True`, `_watch_and_notify` runs per task, polling the store every `poll_interval_ms` and pushing `TaskStatusNotification` on any status change until a terminal state is reached.

### State mapping

All backend-specific state strings are translated once, in [`lifecycle.py`](src/mcp_celery/lifecycle.py):

```python
"PENDING"  → TaskStatus.PENDING
"STARTED"  → TaskStatus.RUNNING
"SUCCESS"  → TaskStatus.COMPLETED
"FAILURE"  → TaskStatus.FAILED
"REVOKED"  → TaskStatus.CANCELLED
# unknown custom states → RUNNING (typically set via update_state)
```

Everything else in the code uses the `TaskStatus` enum.

---

## Public API

```python
from mcp_celery import (
    AsyncToolServer,             # main entry point
    TaskStatus,                  # canonical status enum
    TaskInfo, AsyncToolDef,      # dataclasses
    format_response,             # response builder

    AbstractBackend, CeleryBackend,
    AbstractExposureStrategy, PollingExposureStrategy, OperationResourceStrategy,
    GeneratedTool, ToolRegistry,
    CeleryTaskStore,
    map_celery_state,

    McpCeleryError, TaskNotReady, TaskNotFound, BackendError,
)
```

`AsyncToolServer` is the only class most users touch:

```python
server = AsyncToolServer(
    name="my-server",
    celery_app=celery_app,
    strategy=PollingExposureStrategy(),      # or OperationResourceStrategy(...)
    default_cancel=True,
)

# Decorator form: register-as-you-define
@server.async_tool
def generate_report(user_id: str) -> dict:
    ...

# Imperative form: register an existing Celery task
server.register_async_tool(
    existing_celery_task,
    name="generate_report",
    description="Generate a report for a user",
    cancel=True,
    result_ttl=300,
)

server.run()   # delegates to FastMCP.run()
```

---

## Project layout

```
mcp-celery/
├── pyproject.toml
├── LICENSE
├── CHANGELOG.md
├── README.md
├── src/
│   └── mcp_celery/                    # the importable package
│       ├── __init__.py                # public API (lazy-loads runtime deps)
│       ├── py.typed                   # advertises inline type hints
│       ├── schema.py                  # TaskStatus, AsyncToolDef, TaskInfo, format_response
│       ├── errors.py                  # McpCeleryError, TaskNotReady, TaskNotFound, BackendError
│       ├── lifecycle.py               # Celery state ↔ TaskStatus ↔ MCP status
│       ├── registry.py                # ToolRegistry (name → AsyncToolDef)
│       ├── server.py                  # AsyncToolServer + JSON-schema-from-signature helper
│       ├── backend/
│       │   ├── base.py                # AbstractBackend (dispatch / get_status / get_result / cancel)
│       │   └── celery.py              # CeleryBackend
│       ├── exposure/
│       │   ├── base.py                # AbstractExposureStrategy, GeneratedTool
│       │   ├── polling.py             # PollingExposureStrategy (Approach 1)
│       │   └── operation_resource.py  # OperationResourceStrategy (Approach 3)
│       └── stores/
│           └── celery_task_store.py   # Celery-synced MCP TaskStore (Approach 3)
├── tests/
│   ├── test_tools.py                  # unit tests — no Redis or Celery worker required
│   └── test_operation_resource.py
├── examples/
│   ├── setup.sh                       # one-time venv setup
│   ├── run.sh                         # one-command runner — see Quickstart
│   ├── baseline/                      # manual MCP tools (before)
│   ├── with_package/                  # mcp-celery polling strategy (after)
│   └── approach3/                     # OperationResource strategy + optional notifications
└── docs/
    ├── approach3-design.md            # Approach 3 design notes
    ├── transport-resumable.md         # transport-agnostic resumable requests notes
    └── quickstart-example.py          # stand-alone usage snippet
```

The `src/` layout is deliberate — it prevents `pytest` from silently picking up the in-repo sources instead of the installed wheel, so every test run exercises what users will actually `pip install`.

---

## Running the unit tests

No Redis, no Celery worker needed:

```bash
pip install -e ".[dev]"
pytest tests/test_tools.py -v
```

The tests use a mock backend ([tests/mock_backend.py](tests/mock_backend.py)) and a manual `run(coro)` helper that creates a fresh event loop per call — there are no `pytest-asyncio` markers.
