mcp-worker-sdk

Worker Protocol — Python runtime reference implementation. A universal MCP tool runtime SDK that turns plain Python functions into production-grade, AI-invocable tools.

Write a function, decorate it with @worker.tool, and automatically get schema generation, parameter validation, standardized errors, real bounded concurrency, queue backpressure, and rich runtime observability — with zero boilerplate endpoints.

mcp-worker-sdk is the Python binding of the language-agnostic Worker Protocol. It is Hub-agnostic: omit hub_url and run it standalone as a standard HTTP tool service that any aggregator (Hub / MCP gateway) can consume.

---

✨ Features

• @worker.tool decorator — declare tools as plain functions; the SDK derives the JSON Schema from type hints + docstrings.
• 6 adapters — Shell, DB (7 built-in drivers), Mac GUI, HTTP, MCP client, Custom.
• Zero-boilerplate endpoints — /health, /tools, /execute, /meta auto-assembled via FastAPI.
• Real concurrency & backpressure — shared bounded executor (max_concurrency) + bounded queue (max_queue_length); a full queue replies 429 + Retry-After.
• Rich runtime observability — three-tier /health (status / queue / performance) + health_metrics() returning the protocol HealthMetrics object.
• Lifecycle hooks — on_start / on_health / on_stop / on_error.
• HITL-ready — dangerous operations are kept out of /tools and exposed only as human REST endpoints.
• Standardized errors — eight protocol error codes with HTTP mapping.

---

Installation

pip install mcp-worker-sdk

With all built-in database drivers:

pip install "mcp-worker-sdk[db]"

With resource metrics (psutil, for CPU/memory in /health):

pip install "mcp-worker-sdk[metrics]"

Import name uses an underscore: mcp_worker_sdk.

---

Quick start

# main.py
from mcp_worker_sdk import Worker
from mcp_worker_sdk.adapters import DBAdapter

worker = Worker(
    name="notes",
    adapter=DBAdapter("sqlite", ":memory:"),   # zero-config built-in driver
    # hub_url="https://hub.example.com",        # optional; omit to run standalone
    max_concurrency=10,
    max_queue_length=20,
)


@worker.tool(
    id="notes_search",
    title="Search notes",
    description="Search the notes library by keyword",
    tags=["notes", "search"],
)
def search(keyword: str, limit: int = 10):
    """Search the notes library.

    :param keyword: search keyword
    :param limit: max results to return
    """
    return {"keyword": keyword, "limit": limit}


if __name__ == "__main__":
    worker.run(port=9100)

Verify the auto-generated endpoints:

curl http://localhost:9100/health
curl http://localhost:9100/tools
curl -X POST http://localhost:9100/execute \
  -H "Content-Type: application/json" \
  -d '{"tool_id":"notes_search","params":{"keyword":"mcp"}}'

---

Project structure

Start small with a single file, then split per-domain as the tool count grows:

my-worker/
├── main.py            # entrypoint: register tools + worker.run()
├── worker.py          # Worker singleton (shared by all tool modules)
├── lifecycle.py       # on_start / on_health / on_stop / on_error
├── config.py          # env / config
└── tools/
    ├── __init__.py    # register_all()
    ├── notes.py       # one domain per file
    └── system.py

Since @worker.tool registers at import time, main.py just imports tools.register_all() once. See docs/mcp-worker-sdk/DELIVERY_INTEGRATION.md for the full two-tier layout (single-file vs decoupled).

---

Adapters

Adapter	Purpose
ShellAdapter	Command-line / sandbox / code execution
DBAdapter	Relational / vector / graph / KV / document databases
MacAdapter	macOS GUI / AppleScript / screenshots / mouse & keyboard
HTTPAdapter	Cloud API forwarding (auth, signing hooks, pagination, JSONPath, SSE)
MCPClientAdapter	Wrap a third-party MCP server
CustomAdapter	Fully custom execution

Hardware workers? MacAdapter drives macOS GUI / AppleScript / screenshots; ShellAdapter wraps hardware CLIs (nvidia-smi, sensors, df, kubectl). See the integration guide for real examples.

DBAdapter built-in drivers

db_type	Driver	Capability
postgresql	psycopg	SQL query / connection pool
mysql	pymysql	SQL query / connection pool
sqlite	sqlite3	lightweight SQL (stdlib)
qdrant	qdrant-client	vector search
neo4j	neo4j	graph query (Cypher)
redis	redis	key/value
mongodb	pymongo	document query

---

Lifecycle hooks

@worker.on_start
def on_start():
    connect_db()

@worker.on_health
def on_health():
    return {"db_connected": True}   # merged into /health["custom"]

@worker.on_error
def on_error(exc: Exception):
    logger.error("tool failed", exc_info=exc)

@worker.on_stop
def on_stop():
    disconnect_db()

---

Runtime observability

GET /health returns a rich three-tier payload instead of a bare alive flag:

{
  "status": "busy",
  "degraded_reason": "none",
  "active_tasks": 2,
  "queue_length": 5,
  "max_concurrency": 10,
  "max_queue_length": 20,
  "avg_task_duration_ms": 150,
  "p95_duration_ms": 420,
  "estimated_wait_ms": 250,
  "success_rate": 0.998,
  "cpu_percent": 40.0,
  "memory_percent": 61.2,
  "uptime_seconds": 3600,
  "version": "1.1.0",
  "custom": {"db_connected": true}
}

• status ∈ online / busy / degraded (offline / crashed are derived by the Host Agent / Hub).
• When the queue is full, /execute replies 429 RATE_LIMITED + Retry-After.
• worker.health_metrics() returns a mcp_worker_protocol.HealthMetrics object (5 base + 6 rich fields) for zero-transformation heartbeat aggregation.

---

Error codes

Code	HTTP	Meaning
INVALID_PARAMS	422	Invalid parameters
NOT_FOUND	404	Tool or resource not found
TIMEOUT	504	Execution timeout
PERMISSION_DENIED	403	Not authorized
WORKER_OFFLINE	503	Worker offline
WORKER_ERROR	502	Worker internal error
INTERNAL_ERROR	500	Internal error
RATE_LIMITED	429	Queue full / rate limited

---

HITL (Human-in-the-loop)

Dangerous operations are never registered as MCP tools. Mount them as plain REST endpoints instead:

from mcp_worker_sdk.server import create_app

app = create_app(worker)   # auto-assembles /health /tools /execute /meta

@app.post("/hitl/merge-pr")
def merge_pr(owner: str, repo: str, index: int):
    """Human-only endpoint; never appears in /tools."""
    return gitea.merge(owner, repo, index)

For "same code, two permission surfaces" (e.g. merge visible to business but not auditors), register the tool conditionally via an environment variable — no SDK changes needed.

---

CLI scaffold

mcp-worker create my-worker              # default HTTPAdapter
mcp-worker create my-worker --adapter db

---

Documentation

• Full integration guide (real code + best practices): docs/mcp-worker-sdk/DELIVERY_INTEGRATION.md
• Authoritative spec: docs/mcp-worker-sdk/V1/01-SPEC.md
• Cross-version shared contract: docs/mcp-worker-sdk/shared/00-TERMINOLOGY.md
• Language-agnostic Worker Protocol: docs/worker-protocol/

---

Compatibility

• Python 3.10+
• Runtime dependencies: mcp-worker-protocol>=1.1.0, fastapi, uvicorn, httpx
• Optional: [db] (database drivers), [metrics] (psutil)

---

License

Apache License 2.0