A FastAPI-based Model Context Protocol server for human-in-the-loop workflows
QMCP - Model Context Protocol Server
A spec-aligned Model Context Protocol (MCP) server built with FastAPI.
Features
- ✅ Tool Discovery - List available tools via
/v1/tools - ✅ Tool Invocation - Execute tools via
/v1/tools/{name} - ✅ Invocation History - Audit trail via
/v1/invocations - ✅ Human-in-the-Loop - Request human input via
/v1/human/* - ✅ Persistence - SQLite with SQLModel/aiosqlite
- ✅ Python Client -
qmcp.client.MCPClientfor workflows - ✅ Metaflow Examples - Ready-to-use flow templates
- ✅ Agent Framework - SQLModel schemas + mixins for agent types/topologies
- ✅ PydanticAI Integration - Create agents from QMCP models with full audit trail
- ✅ Structured Logging - JSON logs with structlog
- ✅ Request Tracing - Correlation IDs across requests
- ✅ Metrics - Prometheus-compatible
/metricsendpoint - ✅ CLI Interface - Manage via
qmcpcommand
Quick Start
# Install dependencies
uv sync
# Start the server
uv run qmcp serve
# Or with development reload
uv run qmcp serve --reload
See quickstart.md for a copy-paste walkthrough.
Adoption and Onboarding
Adoption checklist:
- Decide how the server is hosted (local, container, or VM) and who can reach it.
- Set
QMCP_HOST,QMCP_PORT, andQMCP_DATABASE_URLfor your environment. - Standardize
X-Correlation-IDvalues for audit trails across clients. - Decide how humans submit HITL responses (UI or API).
- Wire
/metricsinto your monitoring stack.
Onboarding path:
uv sync --all-extras- Run the end-to-end tutorial below.
uv run qmcp servefor local exploration.
End-to-End Tutorial (HITL approval workflow)
This tutorial mirrors the end-to-end test
tests/test_hitl.py::TestHITLWorkflow::test_complete_approval_workflow.
Copy and paste:
uv sync --all-extras
uv run pytest tests/test_hitl.py::TestHITLWorkflow::test_complete_approval_workflow -v
Client Library
from qmcp.client import MCPClient
with MCPClient(base_url="http://localhost:3333") as client:
# List tools
tools = client.list_tools()
# Invoke a tool
result = client.invoke_tool("echo", {"message": "Hello!"})
print(result.result)
# Human-in-the-loop
request = client.create_human_request(
request_id="approval-001",
request_type="approval",
prompt="Approve deployment?",
options=["approve", "reject"]
)
response = client.wait_for_response("approval-001", timeout=3600)
See docs/client.md for full API documentation.
CLI Commands
# Start the server
qmcp serve [--host HOST] [--port PORT] [--reload]
# Start the server for Docker-based flows
qmcp cookbook serve [--host 0.0.0.0] [--port PORT] [--reload]
# List registered tools
qmcp tools list
# Show configuration
qmcp info
# Run a cookbook flow in Docker
qmcp cookbook simple-plan --goal "Deploy a web service"
# Start the server + run a cookbook flow (unified dev)
qmcp cookbook dev simple-plan --goal "Deploy a web service"
# Run a cookbook flow via the generic runner
qmcp cookbook run simple-plan --goal "Deploy a web service"
# Run other cookbook recipes (flow args are passed through)
qmcp cookbook run approved-deploy --service "api-gateway" --environment "staging"
qmcp cookbook dev local-qc-gauntlet --change-summary "Add audit fields" --target-area "metrics, logging"
# Run a cookbook flow in Docker explicitly
qmcp cookbook docker simple-plan --goal "Deploy a web service"
# If the qmcp shim cannot be installed (Windows)
uv run --no-sync python -m qmcp cookbook run simple-plan --goal "Deploy a web service"
# Run tests with auto setup/teardown
qmcp test [-v] [--coverage] [TEST_PATH]
Cookbook flows run in Docker and require Docker Desktop (Linux engine).
Add --no-sync to skip syncing flow dependencies if the image is already built.
API Endpoints
| Endpoint | Method | Description |
|---|---|---|
/health |
GET | Health check |
/v1/tools |
GET | List available tools |
/v1/tools/{name} |
POST | Invoke a tool |
/v1/invocations |
GET | List invocation history |
/v1/invocations/{id} |
GET | Get single invocation |
/v1/human/requests |
POST | Create human request |
/v1/human/requests |
GET | List human requests |
/v1/human/requests/{id} |
GET | Get request with response |
/v1/human/responses |
POST | Submit human response |
/metrics |
GET | Prometheus metrics |
/metrics/json |
GET | Metrics as JSON |
Built-in Tools
- echo - Echo input back (for testing)
- planner - Create execution plans
- executor - Execute approved plans
- reviewer - Review and assess results
Development
# Install dev dependencies
uv sync --all-extras
# Run tests (with auto cleanup)
uv run qmcp test -v
# Run tests with coverage
uv run qmcp test --coverage
# Run linter
uv run ruff check .
Architecture
See [docs/architectur
Tools (4)
echoEcho input back (for testing)plannerCreate execution plansexecutorExecute approved plansreviewerReview and assess resultsEnvironment Variables
QMCP_HOSTThe host address the server binds toQMCP_PORTThe port the server listens onQMCP_DATABASE_URLSQLite connection string for persistenceConfiguration
{"mcpServers":{"qmcp":{"command":"uv","args":["run","qmcp","serve"],"env":{"QMCP_HOST":"0.0.0.0","QMCP_PORT":"3333","QMCP_DATABASE_URL":"sqlite+aiosqlite:///./qmcp.db"}}}}