mcp-base

Shared base infrastructure package for all MCP servers in the techdeveloper-org ecosystem. Eliminates boilerplate across 13 MCP servers with 4 reusable design patterns.

---

What Is This?

mcp-base is the shared foundation library used by every MCP server in the Claude Workflow Engine ecosystem. Each server repo includes a base/ copy for self-contained deployment.

What it replaces:

• 11 identical _json() serialization functions across server repos
• 109 identical try/except blocks wrapping every tool
• 4 duplicated atomic-write JSON implementations
• 5 duplicated lazy-init client patterns

---

Package Structure

mcp-base/
+-- mcp_base/
|   +-- __init__.py          # Public API re-exports
|   +-- response.py          # MCPResponse builder + success/error helpers
|   +-- decorators.py        # @mcp_tool_handler decorator
|   +-- persistence.py       # AtomicJsonStore, JsonlAppender, SessionIdResolver
|   +-- clients.py           # LazyClient, GitRepoClient, GitHubApiClient, QdrantManager, EmbeddingManager
+-- mcp_errors.py            # Backward-compat error helpers
+-- input_validator.py       # Input validation (null-byte, length, injection detection)
+-- rate_limiter.py          # Token bucket rate limiter
+-- session_hooks.py         # Python bridge to session-mgr MCP (no subprocess overhead)

---

Design Patterns

Pattern	Component	What It Does
Builder	MCPResponse	Fluent response construction with chained .data(), .message(), .build()
Decorator	@mcp_tool_handler	Wraps every tool with error handling and JSON serialization
Repository	AtomicJsonStore	Crash-safe atomic JSON file reads and writes
Singleton/Lazy	LazyClient	Thread-safe lazy initialization for Git, GitHub, Qdrant, embeddings

---

Installation

As a copied package (all server repos use this approach)

Each MCP server includes a base/ directory that is a copy of mcp_base/:

from base import MCPResponse, mcp_tool_handler, AtomicJsonStore, LazyClient
from base.response import success, error

As a pip-installable package

pip install git+https://github.com/techdeveloper-org/mcp-base.git

from mcp_base import MCPResponse, mcp_tool_handler, AtomicJsonStore

Optional dependencies

Install only what your server needs:

pip install gitpython          # for GitRepoClient
pip install PyGithub           # for GitHubApiClient
pip install qdrant-client      # for QdrantManager
pip install sentence-transformers  # for EmbeddingManager

---

API Reference

response.py — MCPResponse Builder

MCPResponse (Builder Pattern)

Fluent builder for MCP tool JSON responses. All tools return a JSON string.

from base.response import MCPResponse, success, error

# Simple success (most common)
return success(branch="main", commits=5)
# -> {"success": true, "branch": "main", "commits": 5}

# Simple error
return error("File not found", error_type="NOT_FOUND")
# -> {"success": false, "error": "File not found", "error_type": "NOT_FOUND"}

# Fluent builder for complex responses
return (MCPResponse.ok()
    .message("PR created successfully")
    .data("pr_number", 42)
    .data("url", "https://github.com/org/repo/pull/42")
    .timestamp()
    .build())

# Failure builder
return (MCPResponse.fail()
    .error_detail("NOT_FOUND", "Branch does not exist", suggestion="Run git fetch first")
    .build())

Methods:

Method	Description
MCPResponse.ok()	Start a success response
MCPResponse.fail()	Start a failure response
.message(str)	Add a human-readable status message
.data(key, value)	Add a single key-value pair
.merge(dict)	Merge a dictionary into the payload
.error_detail(type, msg, suggestion)	Add structured error details
.timestamp()	Inject current ISO-8601 timestamp
.build()	Serialize to JSON string (terminal operation)
.to_dict()	Return payload as a dict (for internal use)

Convenience functions

from base.response import success, error, to_json

success(key="value")              # {"success": true, "key": "value"}
error("msg", error_type="TYPE")   # {"success": false, "error": "msg", ...}
to_json({"any": "dict"})          # pretty-printed JSON string

---

decorators.py — @mcp_tool_handler

Decorator that wraps MCP tool functions with standardized error handling and JSON serialization.

Before (repeated 109 times across server repos):

@mcp.tool()
def my_tool(arg: str) -> str:
    try:
        result = do_work(arg)
        return json.dumps({"success": True, "result": result})
    except Exception as e:
        return json.dumps({"success": False, "error": str(e)})

After:

from base.decorators import mcp_tool_handler

@mcp.tool()
@mcp_tool_handler
def my_tool(arg: str) -> dict:
    result = do_work(arg)
    return {"result": result}  # auto-wrapped: {"success": true, "result": "..."}

With options:

@mcp.tool()
@mcp_tool_handler(include_traceback=True, log_duration=True)
def my_tool(arg: str) -> dict:
    return {"result": process(arg)}
    # -> {"success": true, "result": "...", "duration_ms": 42}

Parameters:

Parameter	Type	Default	Description
error_types	tuple	(Exception,)	Exception types to catch
include_traceback	bool	False	Include last 500 chars of traceback in error response
log_duration	bool	False	Add duration_ms to every response

@validate_params — required parameter check:

from base.decorators import validate_params

@mcp.tool()
@validate_params("session_id", "branch")
@mcp_tool_handler
def my_tool(session_id: str, branch: str) -> dict:
    ...
    # Returns error if session_id or branch is None

---

persistence.py — File-Based Storage

AtomicJsonStore (Repository Pattern)

Thread-safe, crash-safe JSON file persistence using write-to-temp-then-rename.

from base.persistence import AtomicJsonStore
from pathlib import Path

store = AtomicJsonStore(Path("~/.claude/memory/state.json"))

# Load with default
data = store.load(default={"count": 0})

# Save atomically (write .tmp -> rename, never partial writes)
data["count"] += 1
store.save(data)

# Save with backup (.bak created before overwrite)
store.save(data, backup=True)

# Atomic read-modify-write
store.modify(lambda d: d.update({"last_run": "2026-04-14"}))

# Check existence / delete
if store.exists:
    store.delete()

JsonlAppender (Append-Only Log)

Structured event logging in JSON Lines format:

from base.persistence import JsonlAppender
from pathlib import Path

log = JsonlAppender(Path("~/.claude/logs/tools.jsonl"))

# Append entry (auto-timestamps)
log.append({"tool": "Read", "file": "server.py", "status": "ok"})

# Read all entries
for entry in log.read_all():
    print(entry["tool"], entry["timestamp"])

# Filtered read
errors = log.read_filtered(status="error")
today_entries = log.read_filtered(date="2026-04-14")

# Count without loading all into memory
print(f"Total events: {log.count()}")

SessionIdResolver (Singleton)

Resolves current Claude Code session ID from file sources with 30-second TTL caching:

from base.persistence import SessionIdResolver

resolver = SessionIdResolver()
session_id = resolver.get()
# -> "SESSION-20260414-143000-ABCD" or ""

# Force refresh (bypass cache)
session_id = resolver.get(force_refresh=True)

# Invalidate cache
resolver.invalidate()

# Reset singleton (for testing)
SessionIdResolver.reset()

---

clients.py — Lazy Singleton Clients

LazyClient (Abstract Base)

Thread-safe lazy initialization using double-checked locking:

from base.clients import LazyClient

class MyServiceClient(LazyClient):
    def _initialize(self):
        return SomeExpensiveConnection(os.getenv("SERVICE_URL"))
    
    def _health_check(self):
        return {"ping": self._client.ping()}

# Usage
client = MyServiceClient.instance()
service = client.get()        # initializes on first call
service = client.get_or_raise()  # raises RuntimeError if unavailable

print(client.available)       # True/False
print(client.error)           # Last init error or None
print(client.health_check())  # {"client": "MyServiceClient", "available": true, ...}

# Reset (reconnect or for tests)
client.reset()
LazyClient.reset_all()        # reset all registered singletons

GitRepoClient

from base.clients import GitRepoClient

# Singleton for default repo (current directory)
repo = GitRepoClient.instance().get()

# Specific path (not cached)
repo = GitRepoClient.for_path("/path/to/repo")
print(repo.active_branch)

GitHubApiClient

Auto-resolves token from GITHUB_TOKEN env var or gh auth token:

from base.clients import GitHubApiClient

client = GitHubApiClient.instance()
gh_repo = client.get_repo(".")   # auto-detects from git remote URL
issues = gh_repo.get_issues(state="open")

QdrantManager

Local vector database with auto-collection creation:

from base.clients import QdrantManager

qdrant = QdrantManager.instance().get()
# Collections auto-created: tool_calls, sessions, flow_traces, node_decisions
# Each uses 384-dim Cosine similarity

EmbeddingManager

all-MiniLM-L6-v2 sentence embeddings (384-dim):

from base.clients import EmbeddingManager

embedder = EmbeddingManager.instance()
vector = embedder.embed("commit to main branch")
# -> List[float] of length 384, normalized for cosine similarity

---

input_validator.py — Input Validation

from input_validator import validate_input

# Strips null bytes, enforces length limits, detects prompt injection
safe_input = validate_input(user_text, max_length=4096)

rate_limiter.py — Token Bucket Rate Limiter

from rate_limiter import RateLimiter

limiter = RateLimiter(tokens_per_minute=100)
if limiter.allow(client_id="tool-call"):
    process_request()

Enable via environment variable: ENABLE_RATE_LIMITING=1

session_hooks.py — In-Process Session Bridge

Direct Python import bridge to mcp-session-mgr — avoids subprocess overhead for tight-coupling use cases:

from session_hooks import get_current_session, save_session_event

---

How Server Repos Use This Package

Each of the 13 MCP servers includes a base/ directory (copied from mcp_base/):

mcp-git-ops/
+-- server.py
+-- base/                    # <-- copy of mcp_base/
|   +-- __init__.py
|   +-- response.py
|   +-- decorators.py
|   +-- persistence.py
|   +-- clients.py

This design makes each server repo self-contained — no shared package registry required. To update mcp-base in all servers, copy the updated mcp_base/ directory to each server's base/.

---

Development

# Clone
git clone https://github.com/techdeveloper-org/mcp-base.git
cd mcp-base

# Install dev dependencies
pip install -r requirements.txt

# Run tests
pytest tests/ -v

# Lint
ruff check .

---

Ecosystem

mcp-base is used by all 13 MCP servers in the Claude Workflow Engine:

Server	Tools	Purpose
mcp-session-mgr	14	Session lifecycle management
mcp-git-ops	14	Git operations (branch, commit, push)
mcp-github-api	12	GitHub PR, issue, merge workflows
mcp-policy-enforcement	12	Pipeline policy compliance
mcp-token-optimizer	11	Context window token reduction
mcp-pre-tool-gate	9	Pre-tool policy validation
mcp-post-tool-tracker	9	Post-tool progress tracking
mcp-standards-loader	7	Coding standards hot-reload
mcp-uml-diagram	15	UML diagram generation (13 types)
mcp-drawio-diagram	5	Draw.io editable diagrams
mcp-jira-api	10	Jira issue tracking integration
mcp-jenkins-ci	10	Jenkins CI/CD automation
mcp-figma	10	Figma design-to-code pipeline

Parent project: claude-workflow-engine

---

Contributing

1. Fork the repository
2. Create a feature branch: git checkout -b feat/my-improvement
3. Follow coding standards: ASCII-only in .py files, PEP 8, type hints
4. Add tests for new public API
5. Run ruff check . and pytest tests/ before submitting
6. Open a pull request

See claude-workflow-engine CONTRIBUTING.md for full guidelines.

---

License

MIT License — see LICENSE for details.