resk-mcp · Documentation

API reference

Public imports, decorators, helper functions and the exact wire format accepted by /mcp_secure.

• Version 0.1.4
• Python 3.11+
• Built on mcp FastMCP

Public imports

from resk_mcp import (
    SecureMCPServer,
    create_jwt_token,
    verify_jwt_token,
    validate_request_payload,
    detect_pii,
    detect_prompt_injection,
    TokenBasedContextManager,
)

Everything else lives in submodules and must be imported explicitly:

SecureMCPServer

class SecureMCPServer(FastMCP):
    def __init__(
        self,
        name: str = "SecureMCPServer",
        title: str = "Secure MCP Server",
        **kwargs,           # forwarded to FastMCP
    )

On construction the server:

• keeps the FastMCP instance (or creates a FastAPI app when FastMCP does not expose one) as self.secure_app;
• removes /docs, /redoc, /openapi.json and /docs/oauth2-redirect unless RESK_MCP_DOCS is true;
• creates a TokenBasedContextManager(max_tokens=MAX_TOKEN_PER_REQUEST);
• initialises interactions = {"tools": {}, "resources": {}, "prompts": {}} and start_time;
• registers the /mcp_secure route, the dashboard routes and the slowapi limiter.

Because SecureMCPServer subclasses FastMCP, its other methods and attributes (including run() and the SDK transports) remain available. If you need stdio transport for a local agent, use the FastMCP interface; /mcp_secure is the HTTP security layer.

Tool, resource and prompt decorators

@server.tool(name="calculator/add", description="Adds two numbers")
async def add(a: int, b: int) -> int:
    return a + b

@server.resource(uri="info/version")
async def version() -> dict:
    return {"version": "1.0"}

# Backwards-compatible alias for `uri`:
@server.resource(path_pattern="greeting/{name}")
async def greeting(name: str) -> str:
    return f"Hello, {name}"

@server.prompt(name="explain")
async def explain(topic: str) -> str:
    return f"Explain {topic} to a security engineer."

• tool() accepts name, description, annotations and arbitrary FastMCP keyword arguments.
• resource() requires either uri or path_pattern (the latter is deprecated) and raises ValueError if neither is given. It also takes name, description and mime_type.
• prompt() accepts name, description and FastMCP keyword arguments.
• Each decorator returns the FastMCP-processed callable and seeds the interaction counters. Tools are counted under both the bare name and the tool/-prefixed name.

When combining with schema validation, register the tool with an explicit name and let the validator wrap the function body:

validator = ToolSchemaValidator()

@server.tool(name="calculator/add")
@validator.from_decorator(ADD_SCHEMA)
async def add(a: float, b: float) -> float:
    return a + b

JWT helpers

def create_jwt_token(
    user_id: str,
    secret_key: str | None = None,          # defaults to settings.jwt_secret
    algorithm: str | None = None,           # defaults to settings.jwt_algorithm
    expires_delta_minutes: int | None = None,  # defaults to settings.jwt_expiration_minutes
) -> str

def verify_jwt_token(
    token: str,
    secret_key: str | None = None,
    algorithm: str | None = None,
) -> dict[str, Any]

• The only injected claims are user_id and a UTC exp. Add your own claims by passing a token minted elsewhere.
• verify_jwt_token() raises AuthError("Token has expired") for expired tokens and AuthError("Invalid token: …") for everything else (bad signature, wrong algorithm, malformed).
• Passing secret_key explicitly is intended for tests; in production rely on the settings object so client and server cannot drift apart.

Validation helpers

class MCPRequestValidator(BaseModel):
    method: str
    params: dict[str, Any]
    id: int

def validate_request_payload(data: dict[str, Any]) -> MCPRequestValidator
def detect_pii(data: Any) -> bool
def detect_prompt_injection(data: Any) -> bool

validate_request_payload() raises the underlying Pydantic ValidationError when params is not a dict. The two detectors recurse through dicts and lists; detect_pii inspects values only, detect_prompt_injection inspects keys and values. Both return as soon as one match is found.

Patterns are module-level constants, so a fork or a patch can extend them:

from resk_mcp import validation

validation.PII_PATTERNS["SSN"] = r"\b\d{3}-\d{2}-\d{4}\b"
validation.PROMPT_INJECTION_KEYWORDS.append("reveal your hidden rules")

Token budget manager

class TokenBasedContextManager:
    def __init__(self, max_tokens: int = 4000)

    def is_within_limits(self, params: dict[str, Any]) -> bool
    def get_remaining_tokens(self, current_params_tokens: int) -> int

The estimator divides the character length of the JSON serialisation by 4. chars_per_token_approx is an instance attribute fixed at 4 in the constructor — set manager.chars_per_token_approx = 3 after construction if you need a more conservative estimate.

budget = TokenBasedContextManager(max_tokens=2000)
budget.chars_per_token_approx = 3
assert budget.is_within_limits({"text": "x" * 3000})  # 1000 <= 2000

Schema validation

class ToolSchemaValidator:
    def register_schema(self, tool_name: str, schema: dict) -> None
    def validate_parameters(self, tool_name: str, parameters: dict) -> dict
    def from_decorator(self, schema: dict) -> Callable

def generate_schema_from_model(model_class: type) -> dict

validate_parameters() raises SchemaValidationError (with an errors list) when the payload does not match; on success it returns the parameters. from_decorator() registers the schema under the wrapped function's __name__ — which is why you should pass an explicit name= to @server.tool when the two differ.

from pydantic import BaseModel
from resk_mcp.schema_validation import generate_schema_from_model

class TranslateParams(BaseModel):
    text: str
    target: str = "fr"

schema = generate_schema_from_model(TranslateParams)
validator.register_schema("translate", schema)

Security manager

Optional, requires resk-llm. Constructor flags: embedding_function, embedding_dim, similarity_threshold, use_canary_tokens, enable_heuristic_filter, enable_content_moderation, enable_pii_detection, enable_url_detection, enable_ip_protection, patterns_dir.

The module also exposes SecurityException for raise-on-violation flows and a default_security_manager singleton — do not share the singleton across tenants if you add tenant-specific patterns.

Wire format

Request

POST /mcp_secure
Authorization: Bearer <jwt>
Content-Type: application/json

{
  "id": 1,
  "method": "tool/calculator/add",
  "params": {"a": 2, "b": 3}
}

The method suffix is the registered name, slashes included: a tool declared as @server.tool(name="calculator/add") is called with method: "tool/calculator/add".

Responses

{"id": 1, "result": 5}
{"id": 1, "error": {"code": -33000, "message": "Invalid or expired token"}}
{"id": 1, "error": {"code": -32603, "message": "Error executing tool calculator/add: division by zero"}}

• There is no jsonrpc field in the response; id and result/error only.
• Errors carry an optional data member when the server has extra detail.
• Tool results that the SDK returns as a (content, structured) tuple are unwrapped so the wire format stays stable across SDK versions.
• All application-level responses use HTTP 200, including errors. Transport failures use 403/422/429. See error codes.

Unimplemented methods

resource/<name> and prompt/<name> are accepted by the envelope but execute nothing: they return -33004 UNIMPLEMENTED (or -32601 when the name is unknown). Only tools are callable end to end over /mcp_secure in 0.1.0.