1
{
2
  "mcpServers": {
3
    "local-db": {
4
      "command": "node",
5
      "args": ["./mcp-servers/db-server.js"],
6
      "env": {
7
        "DATABASE_URL": "postgres://user:pass@localhost:5432/mydb"
8
      }
9
    }
10
  }
11
}

1
from mcp.server import Server
2
from mcp.server.sse import SseServerTransport
3

4
app = Server("payment-mcp")
5

6
# SECURITY: Add middleware layer
7
@app.middleware("http")
8
async def auth_middleware(request, call_next):
9
    # Transport-level authentication
10
    api_key = request.headers.get("X-API-Key")
11
    if not api_key or not validate_api_key(api_key):
12
        return JSONResponse(
13
            status_code=401,
14
            content={"error": "Invalid API key"}
15
        )
16

17
    # Rate limiting
18
    client_ip = request.client.host
19
    if await rate_limiter.is_rate_limited(client_ip):
20
        return JSONResponse(
21
            status_code=429,
22
            content={"error": "Rate limit exceeded"}
23
        )
24

25
    return await call_next(request)

1
import websockets
2
import jwt
3

4
async def handle_connection(websocket):
5
    # Authenticate on connection
6
    token = await websocket.recv()  # First message must be auth token
7
    try:
8
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
9
        agent_id = payload["agent_id"]
10
        permissions = payload["permissions"]
11
    except jwt.InvalidTokenError:
12
        await websocket.close(code=4001, reason="Authentication failed")
13
        return
14

15
    # Session-scoped context
16
    session = MCPSession(agent_id, permissions)
17

18
    async for message in websocket:
19
        response = await session.handle_message(message)
20
        await websocket.send(response)
21

22
    # Clean up session on disconnect
23
    session.revoke_credentials()

1
import os
2
from fastapi import FastAPI, Header, HTTPException
3

4
app = FastAPI()
5

6
# Load valid API keys from environment or vault
7
VALID_API_KEYS = set(os.getenv("MCP_API_KEYS", "").split(","))
8

9
@app.post("/mcp/tools/call")
10
async def call_tool(
11
    request: ToolCallRequest,
12
    x_api_key: str = Header(None),
13
):
14
    # Validate API key
15
    if not x_api_key or x_api_key not in VALID_API_KEYS:
16
        raise HTTPException(status_code=401, detail="Invalid API key")
17

18
    # Extract permissions from API key identity
19
    key_id, permissions = get_key_permissions(x_api_key)
20

21
    # Log the call
22
    audit_log(
23
        key_id=key_id,
24
        tool=request.tool_name,
25
        params=request.params,
26
        timestamp=datetime.utcnow(),
27
    )
28

29
    # Execute with scoped permissions
30
    return await execute_tool(request.tool_name, request.params, permissions)

1
from authlib.integrations.starlette_client import OAuth
2

3
oauth = OAuth()
4
oauth.register(
5
    name="mcp-provider",
6
    client_id=os.getenv("OAUTH_CLIENT_ID"),
7
    client_secret=os.getenv("OAUTH_CLIENT_SECRET"),
8
    authorize_url="https://auth.company.com/oauth/authorize",
9
    token_url="https://auth.company.com/oauth/token",
10
    client_kwargs={"scope": "mcp:read mcp:tools:execute"},
11
)
12

13
@app.get("/mcp/auth")
14
async def mcp_auth(request: Request):
15
    # Redirect to OAuth provider
16
    redirect = request.url_for("mcp_callback")
17
    return await oauth.mcp_provider.authorize_redirect(request, redirect)
18

19
@app.get("/mcp/auth/callback")
20
async def mcp_callback(request: Request):
21
    token = await oauth.mcp_provider.authorize_access_token(request)
22
    # Token contains identity, permissions, expiry
23
    # Return to MCP client with access token
24
    return {"access_token": token["access_token"], "token_type": "Bearer"}

1
import jwt
2
from datetime import datetime, timedelta
3

4
class SessionTokenManager:
5
    def __init__(self, secret: str):
6
        self.secret = secret
7

8
    def create_session_token(self, agent_id: str, permissions: list[str], ttl: timedelta = timedelta(hours=1)) -> str:
9
        payload = {
10
            "agent_id": agent_id,
11
            "permissions": permissions,
12
            "iat": datetime.utcnow(),
13
            "exp": datetime.utcnow() + ttl,
14
            "jti": uuid4().hex,  # Unique token ID for revocation
15
        }
16
        return jwt.encode(payload, self.secret, algorithm="HS256")
17

18
    def validate_session_token(self, token: str) -> dict:
19
        try:
20
            payload = jwt.decode(token, self.secret, algorithms=["HS256"])
21
            # Check if revoked
22
            if self.is_revoked(payload["jti"]):
23
                raise SecurityError("Token revoked")
24
            return payload
25
        except jwt.ExpiredSignatureError:
26
            raise SecurityError("Token expired")
27
        except jwt.InvalidTokenError:
28
            raise SecurityError("Invalid token")
29

30
    def revoke_token(self, jti: str):
31
        self.revoked_tokens.add(jti)

1
from pydantic import BaseModel, Field, validator
2
from typing import Literal
3

4
class QueryDatabaseParams(BaseModel):
5
    sql: str = Field(..., description="SQL query to execute")
6
    params: list = Field(default_factory=list, description="Query parameters")
7
    max_rows: int = Field(default=100, le=1000, description="Maximum rows to return")
8

9
    @validator("sql")
10
    def validate_sql(cls, v):
11
        # Only allow SELECT queries
12
        if not v.strip().upper().startswith("SELECT"):
13
            raise ValueError("Only SELECT queries are allowed")
14

15
        # Block dangerous patterns
16
        forbidden = ["pg_sleep", "pg_read_file", "COPY", "CREATE", "DROP", "ALTER", "TRUNCATE"]
17
        for pattern in forbidden:
18
            if pattern.lower() in v.lower():
19
                raise ValueError(f"Forbidden SQL pattern: {pattern}")
20

21
        # Prevent SQL injection in dynamic parts
22
        if ";" in v.rstrip(";"):
23
            raise ValueError("Multiple statements not allowed")
24

25
        return v
26

27
class SendEmailParams(BaseModel):
28
    to: list[str] = Field(..., min_items=1, max_items=10)
29
    subject: str = Field(..., max_length=200)
30
    body: str = Field(..., max_length=10000)
31

32
    @validator("to", each_item=True)
33
    def validate_email(cls, v):
34
        if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', v):
35
            raise ValueError(f"Invalid email: {v}")
36

37
        # Check against allowed domains
38
        allowed_domains = ["company.com", "client.org"]
39
        domain = v.split("@")[1]
40
        if domain not in allowed_domains:
41
            raise ValueError(f"Domain {domain} not allowed for email")
42

43
        return v
44

45
@mcp.tool()
46
async def query_database(params: QueryDatabaseParams) -> dict:
47
    """Query the database with validated parameters."""
48
    # Params are already validated by Pydantic
49
    results = await db.execute(params.sql, params.params, limit=params.max_rows)
50
    return {"rows": results, "count": len(results)}

1
# An agent reads a file and calls:
2
tool_call("send_email", {
3
    "to": "admin@company.com",
4
    "subject": "Important: password reset",
5
    "body": "The attacker-controlled content read from a file said: 'Click this link to reset your password: http://evil.com'"
6
})
7
# The body comes from untrusted content — it's an injection via parameter

1
class ToolParameterSanitizer:
2
    """Sanitize tool parameters that may contain injected content."""
3

4
    def sanitize_body(self, body: str, context: str = "") -> str:
5
        # Strip potential injection patterns
6
        body = re.sub(r'<(script|iframe|object|embed)[^>]*>.*?</\1>', '[blocked]', body, flags=re.DOTALL)
7
        body = re.sub(r'javascript:', '', body, flags=re.IGNORECASE)
8
        body = re.sub(r'on\w+\s*=', '', body, flags=re.IGNORECASE)
9

10
        # Truncate to prevent data exfiltration via response length
11
        max_length = 10000 if "read" not in context else 50000
12
        return body[:max_length]
13

14
    def sanitize_filename(self, filename: str) -> str:
15
        # Prevent path traversal
16
        filename = os.path.basename(filename)  # Strip directory components
17
        filename = re.sub(r'[<>:"/\\|?*]', '_', filename)  # Remove dangerous chars
18
        return filename

1
class CostAttackPrevention:
2
    """Detect and prevent cost attacks on MCP servers."""
3

4
    # Tools with high per-call cost
5
    EXPENSIVE_TOOLS = {
6
        "web_search": 0.05,       # $0.05 per call
7
        "llm_completion": 0.10,   # $0.10 per call
8
        "code_generation": 0.15,  # $0.15 per call
9
        "image_generation": 0.05, # $0.05 per call
10
    }
11

12
    BUDGET_LIMITS = {
13
        "per_request": 1.00,      # Max $1 per request
14
        "per_hour": 10.00,        # Max $10 per hour per agent
15
        "per_day": 50.00,         # Max $50 per day per agent
16
    }
17

18
    async def check_budget(self, agent_id: str, tool_name: str) -> bool:
19
        if tool_name not in self.EXPENSIVE_TOOLS:
20
            return True  # Cheap tools always allowed
21

22
        tool_cost = self.EXPENSIVE_TOOLS[tool_name]
23

24
        # Check if this call would exceed per-request budget
25
        if tool_cost > self.BUDGET_LIMITS["per_request"]:
26
            return False
27

28
        # Check running totals
29
        hourly = await self.get_hourly_cost(agent_id)
30
        if hourly + tool_cost > self.BUDGET_LIMITS["per_hour"]:
31
            return False
32

33
        daily = await self.get_daily_cost(agent_id)
34
        if daily + tool_cost > self.BUDGET_LIMITS["per_day"]:
35
            return False
36

37
        return True