ZenGuard MCP Server

A Model Context Protocol (MCP) server that provides access to ZenGuard AI services for detecting Personally Identifiable Information (PII) and prompt injection attempts.

Features

🔍 PII Detection: Identify and flag personally identifiable information in text
🛡️ Prompt Injection Detection: Detect attempts to manipulate AI systems through prompt injection
🔐 Secrets Detection: Detect API keys, passwords, tokens, and sensitive credentials in text
📋 Topic Detection: Detect allowed/banned topics in text based on policy
🎯 Keyword Detection: Detect specific keywords defined in text based on policy
🔐 Secure Authentication: API key-based authentication for all requests
🚀 Multiple Transport Support: SSE and Streamable HTTP transport options
📊 Health Monitoring: Built-in health check endpoints for monitoring
🐳 Container Ready: Docker support with optimized images

Requirements

Python 3.13+
uv package manager (recommended)
ZenGuard API key

Quick Start

1. Installation

# Clone the repository
git clone <repository-url>
cd zenguard-mcp-server

# Install dependencies
uv sync

2. Configuration

Set your ZenGuard API key as an environment variable:

export ZENGUARD_API_KEY="your-api-key-here"

3. Running the Server

Development Mode

uv run main.py

Production Mode

# Using Docker
docker build -t zenguard-mcp .
docker run -p 8000:8000 -e ZENGUARD_API_KEY=your-key zenguard-mcp

# Or using uv directly
uv run main.py

The server will start on http://localhost:8000 by default.

Configuration

Environment Variables

Variable	Default	Description
`MCP_TRANSPORT`	`sse`	MCP transport type (`sse` or `streamable-http`)
`SERVER_HOST`	`0.0.0.0`	Server host address
`SERVER_PORT`	`8000`	Server port
`LOG_LEVEL`	`INFO`	Logging level (`DEBUG`, `INFO`, `WARNING`, `ERROR`)

Transport Options

SSE (Server-Sent Events): Default transport, connect via http://localhost:8000/sse
Streamable HTTP: Alternative transport, connect via http://localhost:8000/mcp

API Reference

MCP Tools

`check_pii`

Detects Personally Identifiable Information in text.

Parameters:

prompt (string): Text to analyze for PII

Returns:

Detection results from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_pii", {"prompt": "My email is john.doe@example.com"})

`check_prompt_injection`

Detects prompt injection attempts in text.

Parameters:

prompt (string): Text to analyze for prompt injection

Returns:

Detection results from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_prompt_injection", {"prompt": "Ignore previous instructions..."})

`check_secrets`

Identifies potential secrets, credentials, or sensitive information within text content. Scans for API keys, passwords, tokens, database connection strings, and other confidential data that should not be exposed in text content.

Parameters:

prompt (string): The text content to scan for secrets

Returns:

Detection result from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_secrets", {"prompt": "API_KEY=sk-1234567890abcdef"})

`check_allowed_topics`

Check allowed topics that have been configured by your policy.

Parameters:

prompt (string): The text content to detect allowed topics

Returns:

Detection result from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_allowed_topics", {"prompt": "Let's discuss machine learning algorithms"})

`check_banned_topics`

Check banned topics that have been configured by your policy.

Parameters:

prompt (string): The text content to detect banned topics

Returns:

Detection result from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_banned_topics", {"prompt": "Content that might contain restricted topics"})

`check_keywords`

Check keywords that you have described in the policy.

Parameters:

prompt (string): The text content to scan for specified keywords

Returns:

Detection result from ZenGuard API

Example:

# Using MCP client
result = await client.call_tool("check_keywords", {"prompt": "Text containing specific keywords to monitor"})

Health Check Endpoints

`GET /api/v1/health`

Basic health check endpoint.

Response:

{
  "status": "healthy",
  "timestamp": "2024-01-01T12:00:00.000000",
  "service": "zenguard-mcp-server",
  "version": "0.1.0"
}

`GET /api/v1/health/detailed`

Detailed health check with additional information.

`GET /api/v1/ready`

Readiness check for container orchestration.

Client Integration

Connecting with MCP Client

import asyncio
from mcp.client.stdio import stdio_client

async def main():
    async with stdio_client() as client:
        # For SSE transport
        await client.connect("http://localhost:8000/sse", headers={
            "zenguard-api-key": "your-api-key"
        })
        
        # For streamable HTTP transport
        # await client.connect("http://localhost:8000/mcp", headers={
        #     "zenguard-api-key": "your-api-key"
        # })
        
        # Use the tools
        result = await client.call_tool("check_pii", {
            "prompt": "Contact me at john@example.com"
        })
        print(result)

asyncio.run(main())

Development

Setup Development Environment

# Install development dependencies
uv sync --extra dev

# Install pre-commit hooks
pre-commit install

Running Tests

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=zenguard_mcp

Code Quality

# Format code
uv run black src/ tests/

# Sort imports
uv run isort src/ tests/

# Type checking
uv run mypy src/

# Linting
uv run flake8 src/ tests/

Project Structure

zenguard-mcp-server/
├── src/
│   └── zenguard_mcp/
│       ├── __init__.py          # Package initialization
│       ├── auth.py              # Authentication logic
│       ├── client.py            # HTTP client for ZenGuard API
│       ├── config.py            # Configuration management
│       ├── health.py            # Health check endpoints
│       ├── server.py            # Main server logic
│       ├── validation.py        # Input validation
│       └── types.py             # Shared types
├── tests/                       # Test suite
├── main.py                      # Entry point
├── pyproject.toml              # Project configuration
├── Dockerfile                   # Container configuration
└── README.md                   # This file

Deployment

Docker

# Build image
docker build -t zenguard-mcp .

# Run container
docker run -p 8000:8000 \
  -e ZENGUARD_API_KEY=your-key \
  -e MCP_TRANSPORT=sse \
  zenguard-mcp

Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: zenguard-mcp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: zenguard-mcp
  template:
    metadata:
      labels:
        app: zenguard-mcp
    spec:
      containers:
      - name: zenguard-mcp
        image: zenguard-mcp:latest
        ports:
        - containerPort: 8000
        env:
        - name: ZENGUARD_API_KEY
          valueFrom:
            secretKeyRef:
              name: zenguard-secrets
              key: api-key
        livenessProbe:
          httpGet:
            path: /api/v1/health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /api/v1/ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

Error Handling

The server provides comprehensive error handling:

401 Unauthorized: Invalid or missing API key
429 Too Many Requests: Rate limit exceeded
500 Internal Server Error: Server-side errors
Timeout Errors: Network or API timeouts

All errors are logged with appropriate detail levels.

Contributing

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Development Guidelines

Follow PEP 8 style guidelines
Add type hints to all functions
Write docstrings for all public functions
Add tests for new features
Update documentation as needed

License

This project is licensed under the MIT License - see the LICENSE file for details.

Support

For support and questions:

Create an issue in the GitHub repository
Contact the development team
Check the ZenGuard API documentation

Changelog

v0.1.0

Initial release
PII detection support
Prompt injection detection support
Secrets detection support
Allowed/Banned topics detection support
Keywords detection support
SSE and Streamable HTTP transport
Health check endpoints
Docker support

ZenGuard MCP Server

Features​

Requirements​

Quick Start​

1. Installation​

2. Configuration​

3. Running the Server​

Development Mode​

Production Mode​

Configuration​

Environment Variables​

Transport Options​

API Reference​

MCP Tools​

check_pii​

check_prompt_injection​

check_secrets​

check_allowed_topics​

check_banned_topics​

check_keywords​

Health Check Endpoints​

GET /api/v1/health​

GET /api/v1/health/detailed​

GET /api/v1/ready​

Client Integration​

Connecting with MCP Client​

Development​

Setup Development Environment​

Running Tests​

Code Quality​

Project Structure​

Deployment​

Docker​

Kubernetes​

Error Handling​

Contributing​

Development Guidelines​

License​

Support​

Changelog​

v0.1.0​

Contents

Features

Requirements

Quick Start

1. Installation

2. Configuration

3. Running the Server

Development Mode

Production Mode

Configuration

Environment Variables

Transport Options

API Reference

MCP Tools

`check_pii`

`check_prompt_injection`

`check_secrets`

`check_allowed_topics`

`check_banned_topics`

`check_keywords`

Health Check Endpoints

`GET /api/v1/health`

`GET /api/v1/health/detailed`

`GET /api/v1/ready`

Client Integration

Connecting with MCP Client

Development

Setup Development Environment

Running Tests

Code Quality

Project Structure

Deployment

Docker

Kubernetes

Error Handling

Contributing

Development Guidelines

License

Support

Changelog

v0.1.0