Skip to content

vhspace/mcp-common

BranchesTags

Repository files navigation

mcp-common

CI Release Python 3.12+ License: Apache-2.0 Ruff

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
  • Credential chain — dynamic token resolution with 1Password op:// auto-detection and cross-process kernel keyring caching

Agent conventions

Working on (or building) a vhspace MCP? Read docs/AGENT_CONVENTIONS.md first. It is the canonical answer to "what does mcp-common already do for me, and what's the convention?" — a curated inventory of every mcp_common.* module, the recommended dual-mode pattern, output and error conventions, the audit checklist, common pitfalls, and the versioning policy.

A tightened skill version of the same content lives at src/mcp_common/shared_skills/mcp-common-conventions/SKILL.md; once #95 lands it will be auto-promoted into every downstream MCP's plugin tree.

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

Environment loading (mcp_common.env)

Standardized .env file loading for both MCP servers and companion CLIs. Solves credential mismatches where the MCP server finds credentials but the CLI does not.

from mcp_common import load_env

# Call once at startup, before MCPSettings() or os.environ reads
load_env()

Precedence with override=False (default — safe for production):

  1. Existing shell/container env vars always win.
  2. .env in the search directory fills in any unset vars.
  3. ../.env one level up (workspace root) fills in any remaining unset vars.

Precedence with override=True (local dev, .env is source of truth):

  1. .env values overwrite existing env vars.
  2. ../.env values overwrite existing env vars (loaded first, so repo .env wins).

CLI entry point pattern:

from mcp_common import load_env, setup_logging

def main():
    load_env()
    logger = setup_logging(name="my-cli")
    # credentials now match what the MCP server sees

File-relative search (for CLIs that may run from any directory):

from pathlib import Path
from mcp_common import load_env

def main():
    load_env(search_from=Path(__file__).parent)

Options:

  • override=False (default) — existing env vars take priority; safe for K8s/Docker
  • override=True — .env values replace existing env vars
  • search_from=Path(...) — base directory for .env search (default: cwd())
  • search_paths=[Path(...)] — explicit list of .env files to load
  • env_file=".env.local" — alternative filename to search for

Idempotent: safe to call multiple times; only the first call loads files.

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
  • For single-value tokens (API keys, bearer tokens), see Credential chain below

Credential chain (mcp_common.credential_chain)

Token resolution with TTL caching, 1Password integration, and cross-process kernel keyring caching for short-lived CLI processes.

Quick start

import requests
from mcp_common.credential_chain import (
    CredentialChain, EnvResolver, CachedResolver, ResolvedAuth,
)

chain = CredentialChain([
    CachedResolver(
        inner=EnvResolver("NETBOX_TOKEN"),
        key_name="mcp:netbox-token",
        ttl_seconds=1800,
    )
], name="netbox")

session = requests.Session()
session.auth = ResolvedAuth(chain, header_format="Token {}")

URI scheme auto-detection

EnvResolver reads the env var value and dispatches based on prefix:

Value Backend Behavior
Plain string Static Used as-is
op://Vault/Item/field 1Password Resolved via op read
vault://... OpenBao Reserved (raises NotImplementedError)

Cross-process caching

CachedResolver stores resolved credentials in the Linux kernel keyring (keyctl):

  • Kernel-enforced TTL (default 30 min)
  • Memory-only (never touches disk)
  • No daemon — shared across all CLIs and MCP servers in the same session
  • One Touch ID prompt covers the entire agent swarm

Setup

For 1Password integration, see docs/credential-chain-setup.md. The setup doc covers:

  • Devcontainer (Linux) via op-forward
  • Native macOS via op desktop integration
  • Headless/CI via service account tokens

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.

Noisy third-party loggers: setup_logging calls suppress_noisy_loggers() by default, which sets urllib3, httpx, requests, and httpcore to WARNING so request lifecycle chatter does not bury application logs. Skipped automatically when level="DEBUG". Opt out with setup_logging(suppress_noisy=False), or call suppress_noisy_loggers(level=..., names=(...)) directly to target a custom set.

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)

CLI helpers (mcp_common.cli)

Shared scaffolding for the companion CLIs that ship alongside each MCP server. Collapses the ~30 lines of bootstrap + custom output + custom typo group repeated across every vhspace MCP into a few imports.

from mcp_common.cli import (
    JsonOption, PaginatedFormatter,
    create_cli_app, echo_result, poll_until, run_cli,
)

app = create_cli_app(
    "netbox-cli",
    project_repo="vhspace/netbox-mcp",
    help="NetBox lookup and search CLI.",
)

@app.command()
def lookup(hostname: str, json: JsonOption = False) -> None:
    result = client.find_device(hostname)
    echo_result(result, as_json=json, human_formatter=device_summary)

def main() -> None:
    run_cli(app, log_name="netbox_cli")

What it gives you:

  • create_cli_app(name, project_repo, **typer_kwargs) — Typer app with no_args_is_help=True, SuggestingTyperGroup as the default group class, and install_cli_exception_handler attached so unhandled exceptions print the agent remediation footer scoped to project_repo.
  • run_cli(app, *, log_name, log_level=None) — chains load_env() → setup_logging(name=log_name, level=…) → app() so every CLI bootstraps consistently.
  • SuggestingTyperGroup — Typer group that emits multi-suggestion Did you mean: 'foo', 'bar'? output for typo'd subcommands; configurable cutoff and max_suggestions via the with_options() factory.
  • JsonOption — reusable --json / -j typer.Option annotation; pair with echo_result(data, as_json=json, …) to honor the flag uniformly.
  • echo_result(data, *, as_json, human_formatter=None, title=None, truncate=4096) — single output sink. JSON mode pretty-prints; human mode defers to human_formatter (or str()), supports a bolded title, and truncates long bodies with an explicit … (N more chars) indicator.
  • PaginatedFormatter(line_fmt, *, show_count=True) — turns {count, results: [...]} REST responses into multi-line human text; drop-in for echo_result's human_formatter.
  • poll_until(fetch, is_terminal, *, timeout_s=600, interval_s=2, on_tick=None) — sync companion to poll_with_progress for CLI commands that wait on AWX jobs, MAAS commissioning, UFM probes, etc. Raises PollTimeout with elapsed_s and last_value attributes on timeout.

See module docstrings under src/mcp_common/cli/ for the full API.

Dual-mode tools (mcp_common.dual_mode)

The headline capability of mcp-common. A single function definition becomes both a FastMCP tool and a Typer CLI command — eliminating the parallel-implementation pattern that duplicated ~500–2000+ LOC across every vhspace MCP CLI.

from fastmcp import FastMCP
from mcp_common.cli import run_cli
from mcp_common.dual_mode import build_cli_from_mcp, dual_mode_tool

mcp = FastMCP("netbox-mcp")

@dual_mode_tool(mcp, cli_name="lookup-device")
def lookup_device(hostname: str, include_interfaces: bool = False) -> dict:
    """Resolve a hostname/IP to a NetBox device."""
    return netbox_client.find_device(hostname, include_interfaces=include_interfaces)

# Same function is now:
#   * a FastMCP tool: lookup_device(hostname="sw01")
#   * a CLI command:  netbox-cli lookup-device --hostname sw01 [--json]

app = build_cli_from_mcp(mcp, project_repo="vhspace/netbox-mcp")

if __name__ == "__main__":
    run_cli(app, log_name="netbox_cli")

What it gives you:

  • @dual_mode_tool(mcp, *, name=None, cli_name=None, cli_group=None, formatters=None, cli_only=False, mcp_only=False, summary=None, **mcp_tool_kwargs) — registers a function as both a FastMCP tool and a deferred Typer CLI command. The MCP namespace prefix is stripped from the CLI command name (so netbox_lookup_device on FastMCP("netbox") becomes lookup-device). Extra mcp_tool_kwargs (annotations, tags, output_schema, …) are forwarded to mcp.tool(...).
  • build_cli_from_mcp(mcp, *, project_repo, name=None, help=None, **typer_kwargs) -> typer.Typer — materializes a Typer CLI from the per-mcp registry. Built on top of create_cli_app, so no_args_is_help, SuggestingTyperGroup, and the agent remediation footer are wired automatically.
  • CliContext — minimal stand-in for fastmcp.Context for CLI runs. ctx.info / ctx.warning / ctx.error / ctx.debug / ctx.log map to the standard logger; ctx.report_progress(progress, total, message) emits a [NN%] message line to stderr. Other Context methods raise AttributeError (discoverable rather than silently no-op).

Parameter introspection covers str, int, float, bool, pathlib.Path, Optional[T], list[T], Literal[...], and Pydantic models. Models with ≤ 6 fields are flattened into individual options (--payload-name, --payload-count, …); larger models fall back to a single --<param>-params '<json>' blob. Async tools are driven by asyncio.run; sync tools call through directly.

Output routes through echo_result, so --json / -j works the same way on every dual-mode CLI command. Pydantic return values serialize via model_dump(mode="json") with sort_keys=True.

Escape hatches:

  • cli_only=True — skip the mcp.tool(...) registration.
  • mcp_only=True — skip the CLI materialization; the tool is registered with FastMCP normally and build_cli_from_mcp filters it out.
  • cli_group="devices" — register the command under a Typer subgroup (netbox-cli devices lookup-device ...).
  • formatters={dict: my_fmt} — per-tool human-mode formatter, passed to echo_result as human_formatter.

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

About

Shared library for Together AI MCP servers — logging, settings, HTTP transport, auth

Topics

python mcp shared-library together-ai model-context-protocol fastmcp

Resources

Readme

License

Apache-2.0 license
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases 27

v0.26.0 Latest
May 29, 2026
+ 26 releases

Packages

 
 
 

Contributors

Languages