安全MCP代码网关模式

# 1. Deploy the pattern
./pattern.sh make install

# 2. Create an API key (recommended)
./scripts/create-api-key.sh alice mcp-log-analyst never

# 3. Add to your AI (.your AI/mcp.json)
{
  "mcpServers": {
    "secure-mcp-gateway": {
      "url": "https://your-gateway-url",
      "transport": {
        "type": "http",
        "headers": {
          "Authorization": "Bearer <your-api-key>"
        }
      }
    }
  }
}

# 4. Ask Your AI
"Use log_store to find all errors from the last hour"

┌────────────────────────────────────────────────────────┐
│                    Shared Infrastructure               │
│  ┌─────────────┐              ┌──────────────────────┐ │
│  │  Keycloak   │              │ OpenShift Logging    │ │
│  │   (SSO)     │              │      (Loki)          │ │
│  └─────────────┘              └──────────────────────┘ │
└────────────────────────────────────────────────────────┘
           │
           ▼
┌─────────────────────────────────────────────────────────┐
│          Multi-Tenant MCP Gateway                       │
│  • Authenticates users (API keys, OAuth, demo token)    │
│  • Loads tool schemas on-demand (not all upfront)       │
│  • Routes requests to appropriate sandboxes             │
│  • Enforces RBAC (users see only authorized tools)      │
└─────────────────────────────────────────────────────────┘
           │
           ├─────────────────┬──────────────────┬
           ▼                 ▼                  ▼
   ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
   │  Sandbox:    │  │  Sandbox:    │  │  Sandbox:    │
   │ Log Analysis │  │   Custom     │  │  YOUR-SERVER │
   │              │  │   Tools      │  │              │
   │  Tools:      │  │  Tools:      │  │  Tools:      │
   │ log_store.py │  │ your-api.py  │  │ ...          │
   │ privacy.py   │  │ ...          │  │              │
   │              │  │              │  │              │
   │ Data flows   │  │ Data flows   │  │ Data flows   │
   │ between      │  │ between      │  │ between      │
   │ tools here   │  │ tools here   │  │ tools here   │
   │ (not via AI) │  │ (not via AI) │  │ (not via AI) │
   └──────────────┘  └──────────────┘  └──────────────┘

# All processing happens in sandbox - data never enters AI context
logs = log_store.query(service="api", level="ERROR", hours=24)
filtered = [log for log in logs if "timeout" in log.message]
summary = privacy.scrub_pii(filtered[:10])
return summary  # Only this small summary goes to AI

# First request: Start processing
dataset = large_data_source.fetch()  # 10GB dataset
progress = process_batch(dataset, batch_size=1000)
save_checkpoint(progress)
return "Processed 1,000 records, checkpoint saved"

# Follow-up request: Continue from checkpoint
progress = load_checkpoint()
result = process_remaining(progress)
return final_summary(result)

# 1. Create your tools
mkdir -p tools/database
cat > tools/database/postgres.py <<'EOF'
def execute_query(database: str, query: str):
    """Execute SQL query against database."""
    return results
EOF

# 2. Clone sandbox chart
cp -r charts/hub/mcp-log-analysis-sandbox charts/hub/mcp-database-sandbox

# 3. Update gateway configuration
# Edit charts/hub/mcp-gateway/values.yaml to add tool set

# 4. Deploy
git add .
git commit -m "Add database tool set"
git push
# ArgoCD automatically syncs

Developer: "Find production errors from last hour"
AI: "Please download logs and paste them..."
Developer: *downloads 500MB of logs*
Developer: *tries to paste*
AI: "Error: Input exceeds context limit"
Developer: *manually filters logs*
Developer: *pastes filtered logs*
AI: "Here are the errors..." (processes 50K tokens)

→ Result: 30 minutes, $5 in tokens, manual work required

Developer: "Find production errors from last hour"
AI: *calls log_store tool in sandbox*
   Sandbox: Queries 2M log lines locally
   Sandbox: Filters for level=ERROR, time=last_hour
   Sandbox: Returns top 5 with context
AI: "Here are the top 5 errors..." (processes 200 tokens)

→ Result: 5 seconds, $0.01 in tokens, fully automated

Request: "Import customer contacts from Sheet-123 to Salesforce"

Step 1: Fetch spreadsheet
  API: Returns 10,000 rows
  Tokens: 150,000 (all rows flow through AI context)

Step 2: AI processes each row
  AI: Validates 10,000 rows in context
  Tokens: 150,000 (reading again)

Step 3: AI writes to CRM
  AI: Formats 10,000 records
  Tokens: 200,000 (writing out)

Total: 500,000 tokens, 5 minutes, $25

Request: "Import customer contacts from Sheet-123 to Salesforce"

Step 1: Gateway routes to sandbox
  Sandbox: Fetches Sheet-123 (10,000 rows stay in sandbox)

Step 2: Sandbox processes internally
  for row in sheet.rows:
    salesforce.create_contact(row)

Step 3: Sandbox returns summary
  Returns: "Imported 10,000 contacts, 3 validation errors"

AI receives: Summary only (50 tokens)

Total: 50 tokens, 30 seconds, $0.02

1. Load sales.csv (50K rows) → 200K tokens in context
2. AI groups by region → keeps all data in context
3. AI calculates top products → still all in context
4. AI formats report → reads everything again
Total: 600K+ tokens

1. Sandbox loads sales.csv (data stays in sandbox)
2. Sandbox runs: df.groupby('region').agg({'sales': 'sum'})
3. Sandbox calculates: df.groupby('product').head(10)
4. Sandbox generates: summary.json with 20 lines
5. AI receives: Only the 20-line summary → 300 tokens
Total: 300 tokens

# Create API key (never expires)
./scripts/create-api-key.sh alice mcp-log-analyst never

# List all API keys
./scripts/list-api-keys.sh

# Revoke API key
./scripts/revoke-api-key.sh <key-id>

# Rotate API key
./scripts/rotate-api-key.sh <key-id>

# Test API keys
./scripts/test-api-keys.sh