| name | mcp-server |
| description | MCP server design, development, and audit using the Python SDK (FastMCP). Use when: creating a new MCP server, wrapping an API as MCP tools, auditing an existing MCP server for patterns and gaps, adding auth/secrets/retry/observability to an MCP server, choosing between tools vs resources vs prompts, designing MCP resources or resource templates, creating MCP prompts for workflow bootstrapping, choosing between low-level Server and FastMCP, designing tool contracts, configuring pydantic-settings, adding structured logging, writing MCP server tests, or improving an existing MCP server codebase. |
| argument-hint | Describe the MCP server to create, audit, or improve |
| tags | ["developer"] |
MCP Server Standards
When to Use
- Creating a new MCP server (wrapping an API, database, or internal service)
- Auditing an existing MCP server for architecture, security, testing, or documentation gaps
- Adding resilience patterns (retry, rate limiting, circuit breaker) to an MCP server
- Adding authentication and secrets management
- Writing or improving MCP server tests
- Designing tool contracts (inputs, outputs, docstrings)
- Setting up configuration, logging, and observability
Framework
Use FastMCP from the official MCP Python SDK (mcp[cli]>=1.27.0). FastMCP provides decorator-based tool registration, automatic input validation from type hints, and built-in lifespan management.
Do not use the low-level mcp.server.Server class unless you need custom protocol handling (e.g., streaming, custom transport). FastMCP covers all standard use cases.
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("my-server", lifespan=app_lifespan)
@mcp.tool()
async def my_tool(
query: str,
limit: int = 25,
ctx: Context = None,
) -> dict[str, Any]:
"""One-sentence description of what this tool does.
Args:
query: What to search for.
limit: Maximum results to return (default 25, max 500).
"""
app = ctx.request_context.lifespan_context
return await app.client.request("GET", "/endpoint", params={"q": query, "limit": limit})
Project Structure
my-mcp-server/
├── src/my_mcp_server/
│ ├── __main__.py # Entry point: mcp.run()
│ ├── server.py # FastMCP instance, lifespan, AppContext
│ ├── client.py # HTTP client with retry, rate limiting, logging
│ ├── auth.py # Pluggable auth via Protocol
│ ├── secrets.py # Keychain / AWS / env secret backends
│ ├── config.py # pydantic-settings BaseSettings
│ ├── observability.py # Structured logging, secret redaction
│ ├── metrics.py # Optional Prometheus instrumentation
│ └── tools/ # 1 file per domain when >10 tools or multiple domains
├── tests/
│ ├── conftest.py
│ ├── unit/ # test_server, test_client, test_auth, test_secrets, test_config
│ └── integration/ # @pytest.mark.integration, skipped by default
├── docs/ # DESIGN.md, ROADMAP.md, SECURITY.md, changes/
├── configs/mcp/ # MCP client config template
├── pyproject.toml
├── README.md
├── ONBOARDING.md
└── uv.lock
Rules
- Use src/ layout with hatchling build system.
- Keep
server.py focused on FastMCP instance, lifespan, and AppContext — not tool logic.
- Each concern (auth, secrets, config, client, logging) gets its own module.
pyproject.toml
[project]
name = "my-mcp-server"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"mcp[cli]>=1.27.0",
"httpx>=0.28",
"pydantic>=2.0",
"pydantic-settings>=2.0",
"structlog>=24.0",
"tenacity>=8.0",
"keyring>=25.0",
]
[dependency-groups]
dev = [
"pytest>=8.0", "pytest-asyncio>=0.24", "pytest-cov>=5.0",
"respx>=0.22", "ruff>=0.9", "mypy>=1.15",
"boto3>=1.34", "boto3-stubs[secretsmanager]>=1.34",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
my-mcp-server = "my_mcp_server.__main__:main"
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]
[tool.ruff.format]
quote-style = "double"
[tool.mypy]
strict = true
python_version = "3.12"
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
markers = ["integration: requires valid API credentials"]
addopts = "-m 'not integration'"
[tool.coverage.report]
fail_under = 85
show_missing = true
exclude_also = ["if TYPE_CHECKING:", "raise NotImplementedError"]
Configuration
Use pydantic-settings BaseSettings with an env prefix unique to the server.
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="MY_SERVER_", env_file=".env", extra="ignore")
base_url: str = "https://api.example.com"
request_timeout: int = 30
auth_strategy: str = "oauth2"
secret_backend: str = "keychain"
max_retries: int = 3
backoff_base: float = 1.0
backoff_max: float = 60.0
rate_limit_rps: int = 10
default_limit: int = 25
max_limit: int = 500
log_level: str = "INFO"
user: str = os.environ.get("USER", "unknown")
Rules: All fields must have sensible defaults. Safety limits (max_limit) must be defined and enforced in tools. Document every env var in the README configuration table.
Authentication
Define a Protocol. Implement concrete strategies. Select via factory.
class AuthManager(Protocol):
async def get_token(self, http_client: httpx.AsyncClient) -> str: ...
async def get_auth_headers(self, http_client: httpx.AsyncClient) -> dict[str, str]: ...
def invalidate(self) -> None: ...
Rules:
- Cache tokens in memory with a 5-minute expiry buffer.
- Use
asyncio.Lock for token refresh to prevent concurrent duplicate refreshes.
- On 401, invalidate the token and retry once (let the retry loop handle it).
- Never log tokens. Exclude from
__repr__ with repr=False.
- Support at least two strategies: one for local dev and one for testing (static token / env var).
Secrets Management
Three-tier strategy: macOS Keychain → AWS Secrets Manager → environment variables.
class SecretStore(Protocol):
def get_secret(self, key: str) -> str: ...
def set_secret(self, key: str, value: str) -> None: ...
Rules:
- Use
keyring for macOS Keychain. Default service name: the package name.
- Use
boto3 for AWS Secrets Manager. Cache the JSON blob in memory after first load.
- Use env vars as fallback for CI/testing.
- Factory function selects backend:
create_secret_store(backend: str) -> SecretStore.
- Never log secret values. Log key names at debug level only.
HTTP Client
Use httpx (async) with retry, rate limiting, and structured logging.
Retry
Use tenacity. Do not implement custom retry loops.
from tenacity import AsyncRetrying, retry_if_exception_type, stop_after_attempt, wait_exponential_jitter
class RetryableHTTPError(Exception):
def __init__(self, status_code: int, retry_after: float | None = None):
self.status_code = status_code
self.retry_after = retry_after
Retryable vs non-retryable:
| Status | Retryable | Action |
|---|
| 429 | Yes | Respect Retry-After header (supports both seconds and HTTP-date format), then backoff |
| 5xx | Yes | Exponential backoff with jitter |
| 401 | Yes (once) | Invalidate token, retry |
| 400, 403, 404 | No | Raise immediately |
| Timeout | No | Raise immediately (don't mask slow upstream) |
Rate Limiting
Token-bucket algorithm with burst support and asyncio.Lock. Tracks _tokens, _max_tokens, _last_refill — refill on each acquire call, sleep if bucket is empty.
VPN-Aware Error Detection
For internal services, detect VPN disconnection and surface a clear message:
def _is_vpn_error(exc: BaseException, url: str = "") -> bool:
domain_is_internal = any(d in url for d in _VPN_REQUIRED_DOMAINS)
if isinstance(exc, httpx.ConnectError):
if "name or service not known" in str(exc.__cause__).lower():
return domain_is_internal
if isinstance(exc, httpx.ConnectTimeout) and domain_is_internal:
return True
return False
Observability
Structured Logging
Use structlog with JSON output to stderr. Configure with: merge_contextvars, add_log_level, TimeStamper(fmt="iso"), _redact_secrets, JSONRenderer.
Secret Redaction
Apply a redaction processor before JSON rendering:
_SENSITIVE_SUBSTRINGS = {"secret", "password", "token", "authorization", "cookie", "api_key"}
def _redact_secrets(logger: Any, method: str, event_dict: dict[str, Any]) -> dict[str, Any]:
for key in list(event_dict.keys()):
if any(s in key.lower() for s in _SENSITIVE_SUBSTRINGS):
event_dict[key] = "***REDACTED***"
return event_dict
Audit Logging
Every tool call must produce a structured audit log entry with: timestamp, event="tool_call", tool, user, request_id (uuid per call, sent as X-Request-Id), duration_ms (wall clock), upstream_duration_ms (HTTP only, via contextvar), result_count, retry_count, status (success | error | zero_result).
Error Classification
Classify errors and return structured responses to the LLM:
class ErrorType(str, Enum):
VPN = "vpn"
AUTH = "auth"
TIMEOUT = "timeout"
RATE_LIMITED = "rate_limited"
NOT_FOUND = "not_found"
VALIDATION = "validation"
UPSTREAM = "upstream"
Return: {"error": "...", "error_type": "vpn", "suggestion": "Connect to VPN and retry."}
Tool Design
Docstrings
Every tool must have: one-sentence description, Args section (every param with type, default, constraints), return description.
Input Validation
- Use type hints for automatic validation via FastMCP.
- Clamp limits server-side:
limit = min(limit, settings.max_limit).
- Validate enum-like inputs against known values. Return structured validation errors, not raw API errors.
- Use Pydantic models for complex dict inputs (create/update operations).
Return Types
- All tools return
dict[str, Any].
- Strip internal-only fields from upstream API responses before returning.
- Return structured error dicts on failure — never raise raw exceptions to the LLM.
- Include metadata (total count, pagination) when relevant.
Resources and Prompts
MCP defines three server-side primitives. Use the right one:
| Primitive | Initiated by | Always in context? | Use for |
|---|
| Tool | Agent | ✅ descriptions always sent | Actions, queries, APIs |
| Resource | App/client | ❌ fetched on demand | Reference docs, schemas, static context |
| Resource template | App/client | ❌ agent fills URI params | Parameterized reference data |
| Prompt | User | ❌ user invokes via UI | Workflow bootstrapping |
Server instructions | Always | ✅ always sent | ≤200-word server orientation |
Key rule: If the agent should independently decide to call it, use a tool. If the host/user/app controls when it appears, use a resource or prompt.
Resources
URI design: Use a custom scheme per RFC 3986 (myapp://path/to/resource). Keep URIs stable. Use {param} for templates (myapp://docs/{slug}). Concrete resources appear in resources/list; templates appear in resources/templates/list only.
Every resource and template must have a name, description, and MIME type (text/markdown, application/json, etc.). Use annotations={"audience": ["assistant"], "priority": 0.8} to signal relevance. Keep payloads under ~8 KB.
Path traversal prevention — always resolve and validate template params before reading from disk:
BASE_DIR = Path(__file__).parent / "docs"
@mcp.resource("myapp://docs/{slug}")
def get_doc(slug: str) -> str:
path = (BASE_DIR / slug).resolve()
if not str(path).startswith(str(BASE_DIR)):
raise McpError(ErrorCode.INVALID_PARAMS, "Invalid slug")
if not path.exists():
raise McpError(ErrorCode.RESOURCE_NOT_FOUND, f"No doc: {slug}")
return path.read_text()
Error codes: not found → -32002, invalid params → -32602, read failure → -32603.
Resources must never expose credentials, API keys, or internal hostnames.
Prompts
Use prompts for user-initiated workflow starters — not as substitutes for tool docstrings. They appear as named entry points in the client UI.
@mcp.prompt(name="investigate", description="Start a root cause investigation.")
def investigate_prompt(metric: str, tenant: str = "", window: str = "1h") -> list[dict]:
return [{"role": "user", "content": f"Investigate `{metric}` over the last {window}..."}]
Rules: Keep arguments minimal and optional where possible. All params must be str. End with a user turn — this tells the LLM what to do next. Never embed secrets or environment-specific values. Sanitize all argument values before including in message content.
Client Support
| Client | Resources | Prompts |
|---|
| Claude Desktop | ✅ Full | ✅ Full |
| VS Code Copilot | ✅ Full | ✅ Slash commands |
| Cursor | ⚠️ Partial | ⚠️ Partial |
Design for progressive enhancement: core functionality must work via tools alone.
Testing
Stack
- pytest + pytest-asyncio (auto mode)
- respx for httpx mocking
- pytest-cov with 85% minimum threshold
Test Structure
| Test file | Covers |
|---|
test_server.py | Tool dispatch, lifespan, AppContext wiring |
test_client.py | Retry on 5xx/429, 401 token refresh, rate limiting, timeout |
test_auth.py | Token acquisition, caching, expiry, refresh, concurrent lock |
test_secrets.py | All backends (keychain mocked, AWS mocked, env) |
test_config.py | Default values, env override, validation |
test_observability.py | Secret redaction, audit log structure, error classification |
Shared Fixtures (conftest.py)
@pytest.fixture
def settings() -> Settings:
return Settings(
base_url="https://api.test.example.com",
secret_backend="env",
request_timeout=10,
max_retries=2,
rate_limit_rps=100,
user="test-user",
)
Integration tests: marked @pytest.mark.integration, skipped by default, require credentials, use read-only operations only.
Documentation
README.md: one-liner description → quick start (clone, sync, test) → VS Code / Claude config (copy-paste) → tools table → env var config table → dev commands → links to docs/.
ONBOARDING.md: prerequisites → install → credential setup → verification → IDE integration → troubleshooting table.
docs/DESIGN.md: problem statement → goals/non-goals → architecture diagram → module responsibilities → key decisions with rationale → tool contracts → testing strategy.
docs/SECURITY.md: secrets management → network requirements → redaction policy → input validation scope → OWASP alignment.
MCP config template (configs/mcp/<server-name>.json):
{
"name": "<server-name>",
"transport": "stdio",
"command": "uv",
"args": ["--directory", "~/code/<server-name>", "run", "python", "-m", "<package_name>"],
"env": { "<PREFIX>_SECRET_BACKEND": "keychain", "<PREFIX>_LOG_LEVEL": "INFO" }
}
Security Checklist
