mcp-common

Shared utilities and testing infrastructure for Python MCP server projects.

• Shared building blocks — config, logging, health checks, and versioning so MCP servers don't reinvent the wheel
• Production HTTP transport — ASGI app factory with CORS, bearer-token auth, and Kubernetes liveness/readiness probes
• Multi-site connection manager — env-var-driven discovery for MCP servers spanning multiple service instances
• Agent error remediation — structured workflow that tells agents to search, dedupe, and file GitHub issues automatically
• Universal plugin generator — one mcp-plugin.toml produces configs for Cursor, Claude Code, OpenCode, and OpenHands
• Cross-MCP hint registry — typed tool references between servers that break at import time when tools are renamed

Plugin Generator Migration (v0.7+)

mcp-plugin-gen now treats pyproject.toml [project].version as the only version source.

• Do not set version in mcp-plugin.toml; generation fails if present.
• Set release version in pyproject.toml, then run mcp-plugin-gen generate ..
• Repin pre-commit hooks to mcp-common v0.7.0 (or newer) in each MCP repo.

Private Claude marketplace mode

mcp-plugin-gen now supports private marketplace registry artifacts for Claude.

• Add optional marketplace metadata in mcp-plugin.toml:

[marketplace]
categories = ["infrastructure", "operations"]
tags = ["mcp", "private", "claude"]

• Generate a single repo entry:

uv run mcp-plugin-gen registry-entry .

• Generate full plugin outputs (also includes registry entry):

uv run mcp-plugin-gen generate .

• Aggregate many repo entries into one deterministic marketplace file:

uv run mcp-plugin-gen aggregate-marketplace /path/to/entries /path/to/marketplace.json

See Private Claude Marketplace Migration for the template and downstream MCP rollout checklist.

Install

uv add git+https://github.com/vhspace/mcp-common

For testing utilities:

uv add "mcp-common[testing] @ git+https://github.com/vhspace/mcp-common"

What's Included

Configuration (mcp_common.config)

Base settings class built on pydantic-settings with .env file support:

from mcp_common import MCPSettings
from pydantic_settings import SettingsConfigDict

class MySettings(MCPSettings):
    model_config = SettingsConfigDict(env_prefix="MY_SERVER_")
    api_url: str
    api_token: str
    timeout: int = 30

Built-in fields: debug, log_level, log_json, unified logging toggles (log_access, log_transcript, log_http_access, redaction lists, log_request_id_header, …), optional github_repo (owner/name) and issue_tracker_url for agent issue workflow (see Agent remediation below).

Credential provider (mcp_common.credentials)

Reusable username/password resolution with audit-safe metadata for MCP servers.

from mcp_common.credentials import (
    CredentialCandidate,
    UsernamePasswordCredentialProvider,
)

provider = UsernamePasswordCredentialProvider(
    candidates=[
        CredentialCandidate(
            name="ORI",
            user_env="REDFISH_ORI_USER",
            password_env="REDFISH_ORI_PASSWORD",
            user_ref_env="REDFISH_ORI_USER_REF",
            password_ref_env="REDFISH_ORI_PASSWORD_REF",
        ),
    ],
    generic_candidate=CredentialCandidate(
        name="GENERIC",
        user_env="REDFISH_USER",
        password_env="REDFISH_PASSWORD",
        user_ref_env="REDFISH_USER_REF",
        password_ref_env="REDFISH_PASSWORD_REF",
    ),
    site_hint_env="REDFISH_SITE",
)

resolved = provider.resolve(host="192.168.196.97")

Notes:

• *_REF env vars resolve via op read ... (1Password CLI)
• plain env vars remain supported for compatibility
• audit event data never includes secret values

Agent remediation (mcp_common.agent_remediation)

Standard markdown for agents when a tool or CLI raises: search issues → thumbs-up if exact duplicate, comment if new info → else open issue → continue the primary task (prefer handling via a subagent).

from mcp_common import format_agent_exception_remediation
from mcp_common.config import MCPSettings
from pydantic_settings import SettingsConfigDict

class MySettings(MCPSettings):
    model_config = SettingsConfigDict(env_prefix="MY_MCP_")
    # github_repo and issue_tracker_url inherited (optional)

text = format_agent_exception_remediation(
    exception=exc,
    project_repo=settings.github_repo or "myorg/my-mcp",
    issue_tracker_url=settings.issue_tracker_url,
    tool_or_command="my_tool",
    version="1.2.3",
    extra_lines=["MAAS site: central"],
)
# Append to MCP tool error text, log line, or stderr as appropriate.

serverUseInstructions snippet (paste into MCP server metadata):

When a tool raises an error, the handling agent should: (1) use a subagent or
focused pass so the primary task can continue; (2) search this repo's GitHub issues
for the error; (3) if a match exists — add a thumbs-up if it is an exact duplicate,
otherwise comment with new logs/repro/version; (4) if no match — open a new issue;
(5) then continue the primary task. Optional: format_agent_exception_remediation
from mcp_common for consistent markdown (github_repo / issue_tracker_url on MCPSettings).

MCP tool wrapper — catches exceptions and re-raises as ToolError with remediation:

from mcp_common import mcp_remediation_wrapper

@mcp.tool()
@mcp_remediation_wrapper(project_repo="myorg/my-mcp")
async def my_tool(arg: str) -> str:
    ...

Multi-site management (mcp_common.sites)

Generic manager for MCP servers that connect to multiple instances of the same service, discovered from environment variables:

from mcp_common.sites import SiteConfig, SiteManager

class WekaSiteConfig(SiteConfig):
    url: str
    username: str
    password: str
    org: str | None = None

class WekaSiteManager(SiteManager[WekaSiteConfig]):
    env_prefix = "WEKA"

mgr = WekaSiteManager(WekaSiteConfig)
mgr.discover()
cfg = mgr.get_site("prod")  # or mgr.get_site() for default

Environment variable conventions (where PREFIX is env_prefix):

Variable	Purpose
{PREFIX}_{SITE}_URL	Required — triggers auto-discovery of a site
{PREFIX}_{SITE}_{FIELD}	Any field on your SiteConfig subclass
{PREFIX}_SITE_ALIASES_JSON	{"alias": "canonical_site"} mapping
{PREFIX}_DEFAULT_SITE	Which site to return from get_site() with no argument

Logging (mcp_common.logging)

Structured logging with JSON mode for containers. Log lines include a stable log_channel field: app (default), access, transcript, or trace, so routers and LLM pipelines can filter without parsing free text.

Defaults are backward compatible: transcript logging is off, HTTP access middleware is off, and JSON merging only adds fields when you use extra= or channel helpers.

from mcp_common import MCPSettings, setup_logging

settings = MCPSettings()  # subclass with env_prefix in real servers
logger = setup_logging(
    level=settings.log_level,
    json_output=settings.log_json,
    name="my-server",
    system_log=True,
)

System log routing: setup_logging attaches a SysLogHandler by default when a platform syslog socket exists (/dev/log on Linux, /var/run/syslog on macOS). Query with journalctl -t my-server --since "1 hour ago" -o json. Silently skipped when the socket is absent.

Timing telemetry: timed_operation (context manager) and log_timing_event (direct call) emit structured timing events on the access channel with operation, expected_s, actual_s, ok, and timed_out fields. poll_with_progress accepts logger and operation for automatic poll timing.

Remediation wrappers with trace logging: mcp_remediation_wrapper and install_cli_exception_handler accept an optional logger parameter. When provided, a trace-channel event is emitted on exception before raising the error, giving log aggregators structured error context alongside the agent-facing remediation markdown.

HTTP access logging is opt-in so existing deployments do not gain new log volume unexpectedly:

from mcp_common import create_http_app

app = create_http_app(
    mcp,
    settings=settings,  # uses log_http_access, log_request_id_header, trace flags
    access_logger=logger,
)

Redaction and truncation: transcript payloads redact keys whose names match built-in sensitive substrings plus log_redact_key_substrings, and optional log_redact_key_patterns (regex per key). Oversized JSON payloads collapse to a small object with _log_truncated, _original_chars, and preview.

Error fingerprints: compute_error_fingerprint(exc) returns a stable 16-char hex id (type, message head, last traceback frame) for deduping alerts; HTTP-only failures use compute_http_error_fingerprint.

See docs/logging-and-telemetry.md for the full downstream adoption guide including aggregator configuration, querying examples, and copy-pasteable smoke tests.

Health Checks (mcp_common.health)

Standard health check responses:

from mcp_common import health_resource

result = health_resource("my-server", "1.0.0", checks={"db": True})
result.to_dict()
# {"name": "my-server", "version": "1.0.0", "status": "healthy", ...}

Version (mcp_common.version)

Runtime version introspection:

from mcp_common import get_version

version = get_version("my-mcp-server")  # "1.2.3" or "0.0.0-dev"

Progress Polling (mcp_common.progress)

Poll long-running operations with MCP progress notifications:

from mcp_common import OperationStates, poll_with_progress

states = OperationStates(success=["complete"], failure=["error"], in_progress=["running"])
result = await poll_with_progress(ctx, check_fn, "status", states, timeout_s=300)

Testing (mcp_common.testing)

Shared pytest fixtures and assertions for MCP servers:

from mcp_common.testing import mcp_client, assert_tool_exists, assert_tool_success

@pytest.fixture
async def client():
    async for c in mcp_client(app):
        yield c

@pytest.mark.anyio
async def test_tools(client):
    await assert_tool_exists(client, "my_tool")
    result = await assert_tool_success(client, "my_tool", {"arg": "value"})

Development

uv sync --all-groups
uv run ruff check src/ tests/
uv run ruff format --check src/ tests/
uv run mypy src/
uv run pytest -v

Bootstrap / doctor

Use plugin doctor checks before first run:

uv run mcp-plugin-gen doctor .

This validates:

• referenced ${ENV_VAR} placeholders in mcp-plugin.toml server env
• optional 1Password CLI/session readiness (op --version, op whoami)

For devcontainers:

• prefer forwarding host env into container runtime (remoteEnv / ${localEnv:...})
• keep desktop agent socket integration as optional, OS-specific best effort

See Devcontainer + 1Password Secret Bridging for host/container setup details.

License

Apache-2.0