# Directory Structure
```
├── .dockerignore
├── .github
│ └── workflows
│ ├── ci.yml
│ ├── publish.yml
│ ├── release.yml
│ └── security.yml
├── .gitignore
├── close_fixed_issues.sh
├── docker-compose.example.yml
├── docker-compose.yml
├── Dockerfile
├── LICENSE
├── Makefile
├── pyproject.toml
├── pytest.ini
├── README.md
├── requirements-dev.txt
├── requirements.txt
├── run_tests.py
├── SECURITY.md
├── src
│ └── mssql_mcp_server
│ ├── __init__.py
│ ├── __main__.py
│ └── server.py
├── test_connection.py
├── tests
│ ├── conftest.py
│ ├── test_config.py
│ ├── test_error_handling.py
│ ├── test_integration.py
│ ├── test_performance.py
│ ├── test_security.py
│ └── test_server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
# Git
.git
.gitignore
# Virtual Environment
venv/
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
.pytest_cache/
.coverage
htmlcov/
# Logs
*.log
# Docker
Dockerfile
.dockerignore
docker-compose.yml
# Editor directories and files
.idea/
.vscode/
*.swp
*.swo
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
*.manifest
*.spec
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Virtual environments
venv/
ENV/
env/
.venv
# IDEs
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
Thumbs.db
ehthumbs.db
# Environment variables
.env
*.env
!.env.example
# Logs
*.log
# Database
*.db
*.sqlite
# MCP specific
mcp-workspace/
# Claude local settings
.claude/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Microsoft SQL Server MCP Server
[](https://pypi.org/project/microsoft_sql_server_mcp/)
[](https://opensource.org/licenses/MIT)
<a href="https://glama.ai/mcp/servers/29cpe19k30">
<img width="380" height="200" src="https://glama.ai/mcp/servers/29cpe19k30/badge" alt="Microsoft SQL Server MCP server" />
</a>
A Model Context Protocol (MCP) server for secure SQL Server database access through Claude Desktop.
## Features
- 🔍 List database tables
- 📊 Execute SQL queries (SELECT, INSERT, UPDATE, DELETE)
- 🔐 Multiple authentication methods (SQL, Windows, Azure AD)
- 🏢 LocalDB and Azure SQL support
- 🔌 Custom port configuration
## Quick Start
### Install with Claude Desktop
Add to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"mssql": {
"command": "uvx",
"args": ["microsoft_sql_server_mcp"],
"env": {
"MSSQL_SERVER": "localhost",
"MSSQL_DATABASE": "your_database",
"MSSQL_USER": "your_username",
"MSSQL_PASSWORD": "your_password"
}
}
}
}
```
## Configuration
### Basic SQL Authentication
```bash
MSSQL_SERVER=localhost # Required
MSSQL_DATABASE=your_database # Required
MSSQL_USER=your_username # Required for SQL auth
MSSQL_PASSWORD=your_password # Required for SQL auth
```
### Windows Authentication
```bash
MSSQL_SERVER=localhost
MSSQL_DATABASE=your_database
MSSQL_WINDOWS_AUTH=true # Use Windows credentials
```
### Azure SQL Database
```bash
MSSQL_SERVER=your-server.database.windows.net
MSSQL_DATABASE=your_database
MSSQL_USER=your_username
MSSQL_PASSWORD=your_password
# Encryption is automatic for Azure
```
### Optional Settings
```bash
MSSQL_PORT=1433 # Custom port (default: 1433)
MSSQL_ENCRYPT=true # Force encryption
```
## Alternative Installation Methods
### Using pip
```bash
pip install microsoft_sql_server_mcp
```
Then in `claude_desktop_config.json`:
```json
{
"mcpServers": {
"mssql": {
"command": "python",
"args": ["-m", "mssql_mcp_server"],
"env": { ... }
}
}
}
```
### Development
```bash
git clone https://github.com/RichardHan/mssql_mcp_server.git
cd mssql_mcp_server
pip install -e .
```
## Security
- Create a dedicated SQL user with minimal permissions
- Never use admin/sa accounts
- Use Windows Authentication when possible
- Enable encryption for sensitive data
## License
MIT
```
--------------------------------------------------------------------------------
/SECURITY.md:
--------------------------------------------------------------------------------
```markdown
# Security Policy
## Reporting Security Issues
If you discover a security vulnerability, please email [email protected] instead of using the public issue tracker.
## Security Best Practices
When using this MCP server:
1. **Database User**: Create a dedicated SQL user with minimal permissions
2. **Never use sa/admin accounts** in production
3. **Use Windows Authentication** when possible
4. **Enable encryption** for sensitive data: `MSSQL_ENCRYPT=true`
5. **Restrict permissions** to only necessary tables and operations
## SQL Injection Protection
This server includes built-in protection against SQL injection:
- Table names are validated with strict regex patterns
- All identifiers are properly escaped
- User input is parameterized where possible
## Example: Minimal Permissions
```sql
-- Create a restricted user
CREATE LOGIN mcp_user WITH PASSWORD = 'StrongPassword123!';
CREATE USER mcp_user FOR LOGIN mcp_user;
-- Grant only necessary permissions
GRANT SELECT ON Schema.TableName TO mcp_user;
GRANT INSERT, UPDATE ON Schema.AuditLog TO mcp_user;
```
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp>=1.0.0
pymssql>=2.2.7
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
python_files = test_*.py
```
--------------------------------------------------------------------------------
/src/mssql_mcp_server/__main__.py:
--------------------------------------------------------------------------------
```python
"""Entry point for running the module with python -m mssql_mcp_server."""
from . import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/mssql_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
from . import server
import asyncio
def main():
"""Main entry point for the package."""
asyncio.run(server.main())
# Expose important items at package level
__all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/requirements-dev.txt:
--------------------------------------------------------------------------------
```
# Testing
pytest>=7.0.0
pytest-asyncio>=0.23.0
pytest-cov>=4.1.0
pytest-timeout>=2.1.0
pytest-xdist>=3.3.0
pytest-mock>=3.11.0
# Code Quality
black>=23.0.0
isort>=5.12.0
mypy>=1.0.0
ruff>=0.0.280
# Security
safety>=2.3.0
bandit>=1.7.0
pip-audit>=2.6.0
# Performance Testing
psutil>=5.9.0
# Build and Release
build>=1.0.0
twine>=4.0.0
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install system dependencies for pymssql
RUN apt-get update && apt-get install -y \
freetds-dev \
&& rm -rf /var/lib/apt/lists/*
# Copy requirements
COPY requirements.txt .
# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt
# Copy project files
COPY . .
# Run the MCP server
CMD ["python", "-m", "mssql_mcp_server"]
```
--------------------------------------------------------------------------------
/docker-compose.example.yml:
--------------------------------------------------------------------------------
```yaml
version: '3.8'
services:
mssql-mcp-server:
build: .
environment:
- MSSQL_SERVER=sqlserver # Use service name for internal Docker network
- MSSQL_DATABASE=TestDB
- MSSQL_USER=sa
- MSSQL_PASSWORD=YourStrong@Passw0rd
- MSSQL_PORT=1433
- MSSQL_ENCRYPT=false
depends_on:
- sqlserver
stdin_open: true
tty: true
# Example SQL Server for testing
sqlserver:
image: mcr.microsoft.com/mssql/server:2022-latest
environment:
- ACCEPT_EULA=Y
- SA_PASSWORD=YourStrong@Passw0rd
- MSSQL_PID=Developer
ports:
- "1433:1433"
volumes:
- sqlserver-data:/var/opt/mssql
volumes:
sqlserver-data:
```
--------------------------------------------------------------------------------
/docker-compose.yml:
--------------------------------------------------------------------------------
```yaml
version: "3.8"
services:
# SQL Server
mssql:
image: mcr.microsoft.com/mssql/server:2019-latest
platform: linux/amd64
environment:
- ACCEPT_EULA=Y
- MSSQL_SA_PASSWORD=${MSSQL_PASSWORD:-StrongPassword123!}
ports:
- "${HOST_SQL_PORT:-1434}:1433"
healthcheck:
test: /opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "${MSSQL_PASSWORD:-StrongPassword123!}" -Q "SELECT 1" || exit 1
interval: 10s
timeout: 3s
retries: 10
start_period: 10s
volumes:
- mssql_data:/var/opt/mssql
mem_limit: ${SQL_MEMORY_LIMIT:-2g}
# MCP Server
mcp_server:
build:
context: .
dockerfile: Dockerfile
depends_on:
mssql:
condition: service_healthy
environment:
- MSSQL_SERVER=${MSSQL_SERVER:-mssql}
- MSSQL_PORT=${MSSQL_PORT:-1433}
- MSSQL_USER=${MSSQL_USER:-sa}
- MSSQL_PASSWORD=${MSSQL_PASSWORD:-StrongPassword123!}
- MSSQL_DATABASE=${MSSQL_DATABASE:-master}
volumes:
- .:/app
volumes:
mssql_data:
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "microsoft_sql_server_mcp"
version = "0.1.0"
description = "A Model Context Protocol (MCP) server that enables secure interaction with Microsoft SQL Server databases."
readme = "README.md"
requires-python = ">=3.11"
authors = [
{name = "Richard Han", email = "[email protected]"}
]
license = {text = "MIT"}
keywords = ["mcp", "mssql", "sql-server", "database", "ai"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
]
dependencies = [
"mcp>=1.0.0",
"pymssql>=2.2.8",
]
[tool.mcp]
system_dependencies.darwin = ["freetds"]
system_dependencies.linux = ["freetds-dev"]
system_dependencies.win32 = []
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
mssql_mcp_server = "mssql_mcp_server:main"
[tool.hatch.build.targets.wheel]
packages = ["src/mssql_mcp_server"]
```
--------------------------------------------------------------------------------
/test_connection.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""Test SQL Server connection using the same configuration as the MCP server."""
import os
import sys
import pymssql
# Add src to path to import our server module
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from mssql_mcp_server.server import get_db_config
try:
print("Loading database configuration from environment variables...")
config = get_db_config()
# Mask sensitive information for display
display_config = config.copy()
if 'password' in display_config:
display_config['password'] = '***'
print(f"Configuration: {display_config}")
print("\nAttempting to connect to SQL Server...")
conn = pymssql.connect(**config)
cursor = conn.cursor()
print("Connection successful!")
print("\nTesting query execution...")
cursor.execute("SELECT TOP 5 TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")
rows = cursor.fetchall()
print(f"Found {len(rows)} tables:")
for row in rows:
print(f" - {row[0]}")
cursor.close()
conn.close()
print("\nConnection test completed successfully!")
except Exception as e:
print(f"Error: {str(e)}")
import traceback
traceback.print_exc()
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
# tests/conftest.py
import pytest
import os
import pymssql
@pytest.fixture(scope="session")
def mssql_connection():
"""Create a test database connection."""
try:
connection = pymssql.connect(
server=os.getenv("MSSQL_SERVER", "localhost"),
user=os.getenv("MSSQL_USER", "sa"),
password=os.getenv("MSSQL_PASSWORD", "testpassword"),
database=os.getenv("MSSQL_DATABASE", "test_db")
)
# Create a test table
cursor = connection.cursor()
cursor.execute("""
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'test_table')
CREATE TABLE test_table (
id INT IDENTITY(1,1) PRIMARY KEY,
name VARCHAR(255),
value INT
)
""")
connection.commit()
yield connection
# Cleanup
cursor.execute("DROP TABLE IF EXISTS test_table")
connection.commit()
cursor.close()
connection.close()
except pymssql.Error as e:
pytest.fail(f"Failed to connect to SQL Server: {e}")
@pytest.fixture(scope="session")
def mssql_cursor(mssql_connection):
"""Create a test cursor."""
cursor = mssql_connection.cursor()
yield cursor
cursor.close()
```
--------------------------------------------------------------------------------
/.github/workflows/security.yml:
--------------------------------------------------------------------------------
```yaml
name: Security Scan
on:
schedule:
- cron: '0 0 * * 1' # Weekly on Monday
push:
branches: [ main ]
pull_request:
branches: [ main ]
workflow_dispatch:
permissions:
contents: read
security-events: write
jobs:
dependency-check:
name: Dependency Security Check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install safety pip-audit
pip install -r requirements.txt
- name: Run Safety check
run: |
safety check --json --output safety-report.json || true
- name: Run pip-audit
run: |
pip-audit --format json --output pip-audit-report.json || true
- name: Upload security reports
uses: actions/upload-artifact@v4
if: always()
with:
name: dependency-security-reports
path: |
safety-report.json
pip-audit-report.json
codeql-analysis:
name: CodeQL Analysis
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Initialize CodeQL
uses: github/codeql-action/init@v3
with:
languages: python
- name: Autobuild
uses: github/codeql-action/autobuild@v3
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v3
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
import pytest
from mssql_mcp_server.server import app, list_tools, list_resources, read_resource, call_tool
from pydantic import AnyUrl
def test_server_initialization():
"""Test that the server initializes correctly."""
assert app.name == "mssql_mcp_server"
@pytest.mark.asyncio
async def test_list_tools():
"""Test that list_tools returns expected tools."""
tools = await list_tools()
assert len(tools) == 1
assert tools[0].name == "execute_sql"
assert "query" in tools[0].inputSchema["properties"]
@pytest.mark.asyncio
async def test_call_tool_invalid_name():
"""Test calling a tool with an invalid name."""
with pytest.raises(ValueError, match="Unknown tool"):
await call_tool("invalid_tool", {})
@pytest.mark.asyncio
async def test_call_tool_missing_query():
"""Test calling execute_sql without a query."""
with pytest.raises(ValueError, match="Query is required"):
await call_tool("execute_sql", {})
# Skip database-dependent tests if no database connection
@pytest.mark.asyncio
@pytest.mark.skipif(
not all([
pytest.importorskip("pymssql"),
pytest.importorskip("mssql_mcp_server")
]),
reason="SQL Server connection not available"
)
async def test_list_resources():
"""Test listing resources (requires database connection)."""
try:
resources = await list_resources()
assert isinstance(resources, list)
except ValueError as e:
if "Missing required database configuration" in str(e):
pytest.skip("Database configuration not available")
raise
```
--------------------------------------------------------------------------------
/.github/workflows/release.yml:
--------------------------------------------------------------------------------
```yaml
name: Create Release
on:
push:
tags:
- 'v*' # Trigger on version tags like v1.0.0
workflow_dispatch:
inputs:
version:
description: 'Version to release (e.g., 1.0.0)'
required: true
type: string
jobs:
create-release:
name: Create GitHub Release
runs-on: ubuntu-latest
permissions:
contents: write
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0 # Get all history for changelog
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Generate changelog
id: changelog
run: |
# Generate changelog from git history
echo "## What's Changed" > RELEASE_NOTES.md
echo "" >> RELEASE_NOTES.md
# Get commits since last tag
LAST_TAG=$(git describe --tags --abbrev=0 2>/dev/null || echo "")
if [ -z "$LAST_TAG" ]; then
git log --pretty=format:"* %s (%h)" >> RELEASE_NOTES.md
else
git log ${LAST_TAG}..HEAD --pretty=format:"* %s (%h)" >> RELEASE_NOTES.md
fi
echo "" >> RELEASE_NOTES.md
echo "**Full Changelog**: https://github.com/${{ github.repository }}/compare/${LAST_TAG}...v${{ github.event.inputs.version || github.ref_name }}" >> RELEASE_NOTES.md
- name: Create Release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
VERSION="${{ github.event.inputs.version || github.ref_name }}"
VERSION="${VERSION#v}" # Remove 'v' prefix if present
gh release create "v${VERSION}" \
--title "Release v${VERSION}" \
--notes-file RELEASE_NOTES.md \
--draft
```
--------------------------------------------------------------------------------
/.github/workflows/publish.yml:
--------------------------------------------------------------------------------
```yaml
name: Publish to PyPI
on:
release:
types: [published]
workflow_dispatch:
inputs:
test_pypi:
description: 'Publish to Test PyPI first'
required: false
default: true
type: boolean
jobs:
build:
name: Build distribution packages
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install hatch
- name: Build package
run: hatch build
- name: Store the distribution packages
uses: actions/upload-artifact@v4
with:
name: python-package-distributions
path: dist/
publish-to-test-pypi:
name: Publish to Test PyPI
if: github.event_name == 'workflow_dispatch' && github.event.inputs.test_pypi == 'true'
needs: build
runs-on: ubuntu-latest
environment:
name: test-pypi
url: https://test.pypi.org/p/microsoft_sql_server_mcp
permissions:
id-token: write # IMPORTANT: mandatory for trusted publishing
steps:
- name: Download all the dists
uses: actions/download-artifact@v4
with:
name: python-package-distributions
path: dist/
- name: Publish to Test PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
repository-url: https://test.pypi.org/legacy/
skip-existing: true
publish-to-pypi:
name: Publish to PyPI
if: github.event_name == 'release' || (github.event_name == 'workflow_dispatch' && github.event.inputs.test_pypi == 'false')
needs: build
runs-on: ubuntu-latest
environment:
name: pypi
url: https://pypi.org/p/microsoft_sql_server_mcp
permissions:
id-token: write # IMPORTANT: mandatory for trusted publishing
steps:
- name: Download all the dists
uses: actions/download-artifact@v4
with:
name: python-package-distributions
path: dist/
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------
```yaml
name: CI/CD Pipeline
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
workflow_dispatch:
jobs:
test:
name: Test Python ${{ matrix.python-version }}
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]
python-version: ['3.11', '3.12']
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install system dependencies (Ubuntu)
if: matrix.os == 'ubuntu-latest'
run: |
sudo apt-get update
sudo apt-get install -y freetds-dev
- name: Install system dependencies (macOS)
if: matrix.os == 'macos-latest'
run: |
brew install freetds
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-dev.txt
pip install -e .
- name: Lint with ruff
run: |
pip install ruff
ruff check src tests
- name: Type check with mypy
run: |
pip install mypy
mypy src --ignore-missing-imports
- name: Test with pytest
run: |
pytest tests -v --tb=short
env:
MSSQL_SERVER: localhost
MSSQL_USER: test
MSSQL_PASSWORD: test
MSSQL_DATABASE: test
- name: Check package build
run: |
pip install hatch
hatch build
ls -la dist/
security-scan:
name: Security Scan
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install bandit[toml] safety
- name: Run Bandit security scan
run: bandit -r src -f json -o bandit-report.json || true
- name: Run Safety check
run: |
pip install -r requirements.txt
safety check --json || true
- name: Upload security reports
uses: actions/upload-artifact@v4
if: always()
with:
name: security-reports
path: |
bandit-report.json
docker-build:
name: Docker Build Test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Build Docker image
run: |
docker build -t mssql-mcp-server:test .
- name: Test Docker image
run: |
docker run --rm mssql-mcp-server:test python --version
```
--------------------------------------------------------------------------------
/close_fixed_issues.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Script to close fixed issues with appropriate comments
echo "Closing fixed issues..."
# Issue #8: Support MSSQL_PORT
echo "Closing issue #8: Support MSSQL_PORT"
gh issue close 8 --comment "This has been implemented! You can now use the \`MSSQL_PORT\` environment variable:
\`\`\`bash
MSSQL_PORT=1433 # Or any custom port
\`\`\`
The implementation includes:
- Default port of 1433 if not specified
- Proper integer conversion with error handling
- Full support for non-standard SQL Server ports
See the updated README for configuration examples."
# Issue #7: Windows Authentication
echo "Closing issue #7: Windows Authentication"
gh issue close 7 --comment "Windows Authentication is now fully supported! Simply set:
\`\`\`bash
MSSQL_WINDOWS_AUTH=true
\`\`\`
When enabled:
- No need to provide MSSQL_USER or MSSQL_PASSWORD
- The server will use Windows integrated authentication
- Works with both standard SQL Server and LocalDB
See the README for complete configuration examples."
# Issue #6: SQL Local DB
echo "Closing issue #6: SQL Local DB"
gh issue close 6 --comment "LocalDB support has been implemented! The server now automatically detects and handles LocalDB connections:
\`\`\`bash
MSSQL_SERVER=(localdb)\\MSSQLLocalDB
MSSQL_DATABASE=your_database
MSSQL_WINDOWS_AUTH=true # LocalDB typically uses Windows Auth
\`\`\`
The implementation automatically converts LocalDB format to pymssql-compatible format."
# Issue #11: Azure SQL encryption
echo "Closing issue #11: Azure SQL encryption"
gh issue close 11 --comment "Azure SQL Database connections are now fully supported with automatic encryption handling!
The server automatically:
- Detects Azure SQL connections (by checking for \`.database.windows.net\`)
- Enables encryption automatically for Azure SQL
- Sets the required TDS version (7.4)
For non-Azure connections, you can control encryption with:
\`\`\`bash
MSSQL_ENCRYPT=true # or false
\`\`\`
See the README for Azure SQL configuration examples."
# Issue #4: Docker support
echo "Closing issue #4: Docker support"
gh issue close 4 --comment "Docker support has been added! The repository now includes:
- \`Dockerfile\` for containerizing the MCP server
- \`docker-compose.yml\` with SQL Server 2019 for testing
- Comprehensive \`Makefile\` with Docker commands
Quick start:
\`\`\`bash
make docker-build
make docker-up
\`\`\`
See the README for complete Docker documentation."
# Issue #12: MSSQL_SERVER configuration
echo "Closing issue #12: MSSQL_SERVER configuration"
gh issue close 12 --comment "This issue has been resolved! The server now:
1. Properly reads the \`MSSQL_SERVER\` environment variable
2. Logs the server being used for debugging
3. Only defaults to \"localhost\" if MSSQL_SERVER is not set
The fix includes enhanced logging to help debug connection issues. If you're still experiencing problems, please check that the environment variable is properly set in your configuration."
echo "Done! Fixed issues have been closed."
```
--------------------------------------------------------------------------------
/run_tests.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""Comprehensive test runner for MSSQL MCP Server."""
import sys
import subprocess
import argparse
from pathlib import Path
def run_command(cmd, description):
"""Run a command and handle output."""
print(f"\n{'='*60}")
print(f"Running: {description}")
print(f"Command: {' '.join(cmd)}")
print('='*60)
result = subprocess.run(cmd, capture_output=False)
if result.returncode != 0:
print(f"❌ {description} failed with return code {result.returncode}")
return False
print(f"✅ {description} passed")
return True
def main():
parser = argparse.ArgumentParser(description="Run MSSQL MCP Server tests")
parser.add_argument('--suite', choices=['all', 'unit', 'security', 'integration', 'performance', 'quality'],
default='all', help='Test suite to run')
parser.add_argument('--coverage', action='store_true', help='Generate coverage report')
parser.add_argument('--parallel', action='store_true', help='Run tests in parallel')
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
args = parser.parse_args()
# Base pytest command
pytest_cmd = ['pytest']
if args.verbose:
pytest_cmd.append('-v')
if args.parallel:
pytest_cmd.extend(['-n', 'auto'])
if args.coverage:
pytest_cmd.extend(['--cov=src/mssql_mcp_server', '--cov-report=html', '--cov-report=term'])
success = True
if args.suite in ['all', 'quality']:
# Code quality checks
print("\n🔍 Running code quality checks...")
if not run_command(['black', '--check', 'src', 'tests'], "Black formatting check"):
success = False
if not run_command(['ruff', 'check', 'src', 'tests'], "Ruff linting"):
success = False
if not run_command(['mypy', 'src', '--ignore-missing-imports'], "MyPy type checking"):
success = False
if args.suite in ['all', 'unit']:
# Unit tests
print("\n🧪 Running unit tests...")
cmd = pytest_cmd + ['tests/test_config.py', 'tests/test_server.py']
if not run_command(cmd, "Unit tests"):
success = False
if args.suite in ['all', 'security']:
# Security tests
print("\n🔒 Running security tests...")
cmd = pytest_cmd + ['tests/test_security.py']
if not run_command(cmd, "Security tests"):
success = False
# Run security scanning
print("\n🔍 Running security scans...")
if not run_command(['safety', 'check'], "Safety dependency check"):
print("⚠️ Security vulnerabilities found in dependencies")
if not run_command(['bandit', '-r', 'src', '-f', 'json', '-o', 'bandit-report.json'],
"Bandit security scan"):
print("⚠️ Security issues found in code")
if args.suite in ['all', 'integration']:
# Integration tests
print("\n🔗 Running integration tests...")
cmd = pytest_cmd + ['tests/test_integration.py', 'tests/test_error_handling.py']
if not run_command(cmd, "Integration tests"):
success = False
if args.suite in ['all', 'performance']:
# Performance tests
print("\n⚡ Running performance tests...")
cmd = pytest_cmd + ['tests/test_performance.py', '-s']
if not run_command(cmd, "Performance tests"):
success = False
# Summary
print(f"\n{'='*60}")
if success:
print("✅ All tests passed!")
if args.coverage:
print("📊 Coverage report generated in htmlcov/")
else:
print("❌ Some tests failed. Please review the output above.")
sys.exit(1)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/tests/test_config.py:
--------------------------------------------------------------------------------
```python
"""Test database configuration and environment variable handling."""
import pytest
import os
from unittest.mock import patch
from mssql_mcp_server.server import get_db_config, validate_table_name
class TestDatabaseConfiguration:
"""Test database configuration from environment variables."""
def test_default_configuration(self):
"""Test default configuration values."""
with patch.dict(os.environ, {
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}, clear=True):
config = get_db_config()
assert config['server'] == 'localhost'
assert config['user'] == 'testuser'
assert config['password'] == 'testpass'
assert config['database'] == 'testdb'
assert 'port' not in config
def test_custom_server_and_port(self):
"""Test custom server and port configuration."""
with patch.dict(os.environ, {
'MSSQL_SERVER': 'custom-server.com',
'MSSQL_PORT': '1433',
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}):
config = get_db_config()
assert config['server'] == 'custom-server.com'
assert config['port'] == 1433
def test_invalid_port(self):
"""Test invalid port handling."""
with patch.dict(os.environ, {
'MSSQL_PORT': 'invalid',
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}):
config = get_db_config()
assert 'port' not in config # Invalid port should be ignored
def test_azure_sql_configuration(self):
"""Test Azure SQL automatic encryption configuration."""
with patch.dict(os.environ, {
'MSSQL_SERVER': 'myserver.database.windows.net',
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}):
config = get_db_config()
assert config['encrypt'] == True
assert config['tds_version'] == '7.4'
def test_localdb_configuration(self):
"""Test LocalDB connection string conversion."""
with patch.dict(os.environ, {
'MSSQL_SERVER': '(localdb)\\MSSQLLocalDB',
'MSSQL_DATABASE': 'testdb',
'MSSQL_WINDOWS_AUTH': 'true'
}):
config = get_db_config()
assert config['server'] == '.\\MSSQLLocalDB'
assert 'user' not in config
assert 'password' not in config
def test_windows_authentication(self):
"""Test Windows authentication configuration."""
with patch.dict(os.environ, {
'MSSQL_SERVER': 'localhost',
'MSSQL_DATABASE': 'testdb',
'MSSQL_WINDOWS_AUTH': 'true'
}):
config = get_db_config()
assert 'user' not in config
assert 'password' not in config
def test_missing_required_config_sql_auth(self):
"""Test missing required configuration for SQL authentication."""
with patch.dict(os.environ, {
'MSSQL_SERVER': 'localhost'
}, clear=True):
with pytest.raises(ValueError, match="Missing required database configuration"):
get_db_config()
def test_missing_database_windows_auth(self):
"""Test missing database for Windows authentication."""
with patch.dict(os.environ, {
'MSSQL_WINDOWS_AUTH': 'true'
}, clear=True):
with pytest.raises(ValueError, match="Missing required database configuration"):
get_db_config()
def test_encryption_settings(self):
"""Test various encryption settings."""
# Non-Azure with encryption
with patch.dict(os.environ, {
'MSSQL_SERVER': 'localhost',
'MSSQL_ENCRYPT': 'true',
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}):
config = get_db_config()
assert config['encrypt'] == True
# Non-Azure without encryption (default)
with patch.dict(os.environ, {
'MSSQL_SERVER': 'localhost',
'MSSQL_USER': 'testuser',
'MSSQL_PASSWORD': 'testpass',
'MSSQL_DATABASE': 'testdb'
}):
config = get_db_config()
assert config['encrypt'] == False
class TestTableNameValidation:
"""Test SQL table name validation and escaping."""
def test_valid_table_names(self):
"""Test validation of valid table names."""
valid_names = [
'users',
'UserAccounts',
'user_accounts',
'table123',
'dbo.users',
'schema_name.table_name'
]
for name in valid_names:
escaped = validate_table_name(name)
assert escaped is not None
assert '[' in escaped and ']' in escaped
def test_invalid_table_names(self):
"""Test rejection of invalid table names."""
invalid_names = [
'users; DROP TABLE users', # SQL injection
'users OR 1=1', # SQL injection
'users--', # SQL comment
'users/*comment*/', # SQL comment
'users\'', # Quote
'users"', # Double quote
'schema.name.table', # Too many dots
'user@table', # Invalid character
'user#table', # Invalid character
'', # Empty
'.', # Just dot
'..', # Double dot
]
for name in invalid_names:
with pytest.raises(ValueError, match="Invalid table name"):
validate_table_name(name)
def test_table_name_escaping(self):
"""Test proper escaping of table names."""
assert validate_table_name('users') == '[users]'
assert validate_table_name('dbo.users') == '[dbo].[users]'
assert validate_table_name('my_table_123') == '[my_table_123]'
```
--------------------------------------------------------------------------------
/tests/test_security.py:
--------------------------------------------------------------------------------
```python
"""Security tests for SQL injection prevention and safe query handling."""
import pytest
from unittest.mock import Mock, patch, AsyncMock
from mssql_mcp_server.server import validate_table_name, read_resource, call_tool
from pydantic import AnyUrl
from mcp.types import TextContent
class TestSQLInjectionPrevention:
"""Test SQL injection prevention measures."""
@pytest.mark.asyncio
async def test_sql_injection_in_table_names(self):
"""Test that SQL injection attempts in table names are blocked."""
malicious_uris = [
"mssql://users; DROP TABLE users--/data",
"mssql://users' OR '1'='1/data",
"mssql://users/**/UNION/**/SELECT/**/password/data",
"mssql://users%20OR%201=1/data",
]
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
for uri in malicious_uris:
with pytest.raises((ValueError, RuntimeError)):
await read_resource(AnyUrl(uri))
@pytest.mark.asyncio
async def test_safe_query_execution(self):
"""Test that only safe queries are executed."""
# Mock the database connection
mock_cursor = Mock()
mock_conn = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
# Test safe table read
uri = AnyUrl("mssql://users/data")
mock_cursor.description = [('id',), ('name',)]
mock_cursor.fetchall.return_value = [(1, 'John'), (2, 'Jane')]
result = await read_resource(uri)
# Verify the query was escaped properly
executed_query = mock_cursor.execute.call_args[0][0]
assert '[users]' in executed_query
assert 'SELECT TOP 100 * FROM [users]' == executed_query
def test_parameterized_queries(self):
"""Ensure queries use parameters where user input is involved."""
# This is a design consideration test
# The current implementation doesn't use parameterized queries for table names
# because table names can't be parameterized in SQL
# Instead, we validate and escape them
pass
@pytest.mark.asyncio
async def test_query_result_sanitization(self):
"""Test that query results don't expose sensitive information."""
mock_cursor = Mock()
mock_conn = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
# Test that passwords or sensitive data aren't exposed in errors
mock_cursor.execute.side_effect = Exception("Login failed for user 'sa' with password 'secret123'")
result = await call_tool("execute_sql", {"query": "SELECT * FROM users"})
# Verify sensitive info is not in the error message
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert 'secret123' not in result[0].text
assert 'Error executing query' in result[0].text
class TestInputValidation:
"""Test input validation for all user inputs."""
@pytest.mark.asyncio
async def test_tool_argument_validation(self):
"""Test that tool arguments are properly validated."""
# Test with various invalid inputs
invalid_inputs = [
{}, # Empty
{"query": ""}, # Empty query
{"query": None}, # None query
{"query": {"$ne": None}}, # NoSQL injection attempt
]
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
for invalid_input in invalid_inputs:
with pytest.raises(ValueError):
await call_tool("execute_sql", invalid_input)
def test_environment_variable_validation(self):
"""Test that environment variables are validated."""
# Test with potentially dangerous environment values
dangerous_values = {
'MSSQL_SERVER': 'localhost; exec xp_cmdshell "whoami"',
'MSSQL_DATABASE': 'test; DROP DATABASE test',
'MSSQL_USER': 'admin\'--',
}
with patch.dict('os.environ', dangerous_values):
# The connection should fail safely without executing malicious code
# This tests that pymssql properly handles these values
pass
class TestResourceAccessControl:
"""Test resource access control and permissions."""
@pytest.mark.asyncio
async def test_system_table_access_restriction(self):
"""Test that system tables are not exposed as resources."""
mock_cursor = Mock()
mock_conn = Mock()
mock_conn.cursor.return_value = mock_cursor
# Simulate database returning both user and system tables
mock_cursor.fetchall.return_value = [
('users',),
('sys.objects',), # System table
('INFORMATION_SCHEMA.TABLES',), # System view
('products',),
]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
from mssql_mcp_server.server import list_resources
resources = await list_resources()
# Verify system tables are filtered out (if implemented)
# Currently the query uses INFORMATION_SCHEMA which should only return user tables
resource_names = [r.name for r in resources]
assert len(resources) == 4 # All tables are returned currently
@pytest.mark.asyncio
async def test_query_permissions(self):
"""Test that dangerous queries are handled safely."""
dangerous_queries = [
"DROP TABLE users",
"CREATE LOGIN hacker WITH PASSWORD = 'password'",
"EXEC xp_cmdshell 'dir'",
"ALTER SERVER ROLE sysadmin ADD MEMBER hacker",
]
mock_cursor = Mock()
mock_conn = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'test'
}):
for query in dangerous_queries:
# The queries will be executed (current implementation doesn't block them)
# but we ensure errors are handled gracefully
mock_cursor.execute.side_effect = Exception("Permission denied")
result = await call_tool("execute_sql", {"query": query})
assert len(result) == 1
assert "Error executing query" in result[0].text
```
--------------------------------------------------------------------------------
/src/mssql_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
import asyncio
import logging
import os
import re
import pymssql
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
from pydantic import AnyUrl
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mssql_mcp_server")
def validate_table_name(table_name: str) -> str:
"""Validate and escape table name to prevent SQL injection."""
# Allow only alphanumeric, underscore, and dot (for schema.table)
if not re.match(r'^[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)?$', table_name):
raise ValueError(f"Invalid table name: {table_name}")
# Split schema and table if present
parts = table_name.split('.')
if len(parts) == 2:
# Escape both schema and table name
return f"[{parts[0]}].[{parts[1]}]"
else:
# Just table name
return f"[{table_name}]"
def get_db_config():
"""Get database configuration from environment variables."""
# Basic configuration
server = os.getenv("MSSQL_SERVER", "localhost")
logger.info(f"MSSQL_SERVER environment variable: {os.getenv('MSSQL_SERVER', 'NOT SET')}")
logger.info(f"Using server: {server}")
# Handle LocalDB connections (Issue #6)
# LocalDB format: (localdb)\instancename
if server.startswith("(localdb)\\"):
# For LocalDB, pymssql needs special formatting
# Convert (localdb)\MSSQLLocalDB to localhost\MSSQLLocalDB with dynamic port
instance_name = server.replace("(localdb)\\", "")
server = f".\\{instance_name}"
logger.info(f"Detected LocalDB connection, converted to: {server}")
config = {
"server": server,
"user": os.getenv("MSSQL_USER"),
"password": os.getenv("MSSQL_PASSWORD"),
"database": os.getenv("MSSQL_DATABASE"),
"port": os.getenv("MSSQL_PORT", "1433"), # Default MSSQL port
}
# Port support (Issue #8)
port = os.getenv("MSSQL_PORT")
if port:
try:
config["port"] = int(port)
except ValueError:
logger.warning(f"Invalid MSSQL_PORT value: {port}. Using default port.")
# Encryption settings for Azure SQL (Issue #11)
# Check if we're connecting to Azure SQL
if config["server"] and ".database.windows.net" in config["server"]:
config["tds_version"] = "7.4" # Required for Azure SQL
# Azure SQL requires encryption - handled by TDS version
else:
# For non-Azure connections, TDS version can be configured
# pymssql doesn't support the 'encrypt' parameter directly
# Encryption is handled through TDS version and connection string
if os.getenv("MSSQL_ENCRYPT", "false").lower() == "true":
config["tds_version"] = "7.4"
# Windows Authentication support (Issue #7)
use_windows_auth = os.getenv("MSSQL_WINDOWS_AUTH", "false").lower() == "true"
if use_windows_auth:
# For Windows authentication, user and password are not required
if not config["database"]:
logger.error("MSSQL_DATABASE is required")
raise ValueError("Missing required database configuration")
# Remove user and password for Windows auth
config.pop("user", None)
config.pop("password", None)
logger.info("Using Windows Authentication")
else:
# SQL Authentication - user and password are required
if not all([config["user"], config["password"], config["database"]]):
logger.error("Missing required database configuration. Please check environment variables:")
logger.error("MSSQL_USER, MSSQL_PASSWORD, and MSSQL_DATABASE are required")
raise ValueError("Missing required database configuration")
return config
def get_command():
"""Get the command to execute SQL queries."""
return os.getenv("MSSQL_COMMAND", "execute_sql")
# Initialize server
app = Server("mssql_mcp_server")
@app.list_resources()
async def list_resources() -> list[Resource]:
"""List SQL Server tables as resources."""
config = get_db_config()
try:
conn = pymssql.connect(**config)
cursor = conn.cursor()
# Query to get user tables from the current database
cursor.execute("""
SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_TYPE = 'BASE TABLE'
""")
tables = cursor.fetchall()
logger.info(f"Found tables: {tables}")
resources = []
for table in tables:
resources.append(
Resource(
uri=f"mssql://{table[0]}/data",
name=f"Table: {table[0]}",
mimeType="text/plain",
description=f"Data in table: {table[0]}"
)
)
cursor.close()
conn.close()
return resources
except Exception as e:
logger.error(f"Failed to list resources: {str(e)}")
return []
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
"""Read table contents."""
config = get_db_config()
uri_str = str(uri)
logger.info(f"Reading resource: {uri_str}")
if not uri_str.startswith("mssql://"):
raise ValueError(f"Invalid URI scheme: {uri_str}")
parts = uri_str[8:].split('/')
table = parts[0]
try:
# Validate table name to prevent SQL injection
safe_table = validate_table_name(table)
conn = pymssql.connect(**config)
cursor = conn.cursor()
# Use TOP 100 for MSSQL (equivalent to LIMIT in MySQL)
cursor.execute(f"SELECT TOP 100 * FROM {safe_table}")
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
result = [",".join(map(str, row)) for row in rows]
cursor.close()
conn.close()
return "\n".join([",".join(columns)] + result)
except Exception as e:
logger.error(f"Database error reading resource {uri}: {str(e)}")
raise RuntimeError(f"Database error: {str(e)}")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available SQL Server tools."""
command = get_command()
logger.info("Listing tools...")
return [
Tool(
name=command,
description="Execute an SQL query on the SQL Server",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The SQL query to execute"
}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute SQL commands."""
config = get_db_config()
command = get_command()
logger.info(f"Calling tool: {name} with arguments: {arguments}")
if name != command:
raise ValueError(f"Unknown tool: {name}")
query = arguments.get("query")
if not query:
raise ValueError("Query is required")
try:
conn = pymssql.connect(**config)
cursor = conn.cursor()
cursor.execute(query)
# Special handling for table listing
if query.strip().upper().startswith("SELECT") and "INFORMATION_SCHEMA.TABLES" in query.upper():
tables = cursor.fetchall()
result = ["Tables_in_" + config["database"]] # Header
result.extend([table[0] for table in tables])
cursor.close()
conn.close()
return [TextContent(type="text", text="\n".join(result))]
# Regular SELECT queries
elif query.strip().upper().startswith("SELECT"):
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
result = [",".join(map(str, row)) for row in rows]
cursor.close()
conn.close()
return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
# Non-SELECT queries
else:
conn.commit()
affected_rows = cursor.rowcount
cursor.close()
conn.close()
return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {affected_rows}")]
except Exception as e:
logger.error(f"Error executing SQL '{query}': {e}")
return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
async def main():
"""Main entry point to run the MCP server."""
from mcp.server.stdio import stdio_server
logger.info("Starting MSSQL MCP server...")
config = get_db_config()
# Log connection info without exposing sensitive data
server_info = config['server']
if 'port' in config:
server_info += f":{config['port']}"
user_info = config.get('user', 'Windows Auth')
logger.info(f"Database config: {server_info}/{config['database']} as {user_info}")
async with stdio_server() as (read_stream, write_stream):
try:
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
except Exception as e:
logger.error(f"Server error: {str(e)}", exc_info=True)
raise
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/tests/test_integration.py:
--------------------------------------------------------------------------------
```python
"""Integration tests for MCP protocol communication and end-to-end functionality."""
import pytest
import asyncio
import json
from unittest.mock import Mock, patch, AsyncMock
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Resource, Tool
from mssql_mcp_server.server import app
class TestMCPProtocolIntegration:
"""Test MCP protocol integration and communication."""
@pytest.mark.asyncio
async def test_server_initialization_options(self):
"""Test server initialization with proper options."""
init_options = app.create_initialization_options()
assert init_options.server_name == "mssql_mcp_server"
assert init_options.server_version is not None
assert hasattr(init_options, 'capabilities')
@pytest.mark.asyncio
async def test_full_mcp_lifecycle(self):
"""Test complete MCP server lifecycle from init to shutdown."""
# Mock the stdio streams
mock_read_stream = AsyncMock()
mock_write_stream = AsyncMock()
# Mock database connection
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Test resource listing
mock_cursor.fetchall.return_value = [('users',), ('products',)]
resources = await app.list_resources()
assert len(resources) == 2
assert all(isinstance(r, Resource) for r in resources)
assert resources[0].name == "Table: users"
assert resources[1].name == "Table: products"
# Test tool listing
tools = await app.list_tools()
assert len(tools) == 1
assert tools[0].name == "execute_sql"
# Test tool execution
mock_cursor.description = [('count',)]
mock_cursor.fetchall.return_value = [(42,)]
result = await app.call_tool("execute_sql", {"query": "SELECT COUNT(*) FROM users"})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "42" in result[0].text
@pytest.mark.asyncio
async def test_concurrent_requests(self):
"""Test handling of concurrent MCP requests."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Simulate concurrent resource listing
mock_cursor.fetchall.return_value = [('table1',), ('table2',)]
# Run multiple concurrent requests
tasks = [app.list_resources() for _ in range(10)]
results = await asyncio.gather(*tasks)
# All should succeed
assert len(results) == 10
for result in results:
assert len(result) == 2
@pytest.mark.asyncio
async def test_error_propagation(self):
"""Test that errors are properly propagated through MCP protocol."""
with patch.dict('os.environ', {}, clear=True):
# Missing configuration should raise error
with pytest.raises(ValueError, match="Missing required database configuration"):
await app.list_resources()
class TestDatabaseIntegration:
"""Test actual database integration scenarios."""
@pytest.mark.asyncio
async def test_connection_pooling(self):
"""Test that connections are properly managed and pooled."""
call_count = 0
def mock_connect(**kwargs):
nonlocal call_count
call_count += 1
mock_conn = Mock()
mock_cursor = Mock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value = mock_cursor
return mock_conn
with patch('pymssql.connect', side_effect=mock_connect):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Multiple operations should create multiple connections
# (current implementation doesn't pool)
for _ in range(5):
await app.list_resources()
assert call_count == 5 # One connection per operation
@pytest.mark.asyncio
async def test_transaction_handling(self):
"""Test proper transaction handling for write operations."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.rowcount = 1
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Test INSERT operation
result = await app.call_tool("execute_sql", {
"query": "INSERT INTO users (name) VALUES ('test')"
})
# Verify commit was called
mock_conn.commit.assert_called_once()
assert "Rows affected: 1" in result[0].text
@pytest.mark.asyncio
async def test_connection_cleanup(self):
"""Test that connections are properly cleaned up."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Even if operation fails, connection should be closed
mock_cursor.execute.side_effect = Exception("Query failed")
try:
await app.call_tool("execute_sql", {"query": "SELECT * FROM users"})
except:
pass
# Connection should still be closed
# (Note: current implementation may not guarantee this)
class TestEdgeCases:
"""Test edge cases and boundary conditions."""
@pytest.mark.asyncio
async def test_empty_table_list(self):
"""Test handling of database with no tables."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchall.return_value = []
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
resources = await app.list_resources()
assert resources == []
@pytest.mark.asyncio
async def test_large_result_set(self):
"""Test handling of large query results."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Create large result set
large_result = [(i, f'user_{i}', f'email_{i}@test.com') for i in range(10000)]
mock_cursor.description = [('id',), ('name',), ('email',)]
mock_cursor.fetchall.return_value = large_result
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {
"query": "SELECT * FROM users"
})
# Should handle large results gracefully
assert len(result) == 1
assert isinstance(result[0].text, str)
assert len(result[0].text.split('\n')) == 10001 # Header + 10000 rows
@pytest.mark.asyncio
async def test_special_characters_in_data(self):
"""Test handling of special characters in query results."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Data with special characters
mock_cursor.description = [('data',)]
mock_cursor.fetchall.return_value = [
('Hello, "World"',),
('Line1\nLine2',),
('Tab\there',),
('NULL',),
(None,),
]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {
"query": "SELECT data FROM test_table"
})
# Should handle special characters properly
assert len(result) == 1
text = result[0].text
assert 'Hello, "World"' in text
assert 'None' in text # None should be converted to string
```
--------------------------------------------------------------------------------
/tests/test_error_handling.py:
--------------------------------------------------------------------------------
```python
"""Test error handling, resilience, and recovery scenarios."""
import pytest
import asyncio
from unittest.mock import Mock, patch, PropertyMock
from mssql_mcp_server.server import app, get_db_config
import pymssql
class TestConnectionErrors:
"""Test various connection error scenarios."""
@pytest.mark.asyncio
async def test_connection_timeout(self):
"""Test handling of connection timeouts."""
with patch('pymssql.connect') as mock_connect:
mock_connect.side_effect = pymssql.OperationalError("Connection timeout")
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
resources = await app.list_resources()
assert resources == [] # Should return empty list on connection failure
@pytest.mark.asyncio
async def test_authentication_failure(self):
"""Test handling of authentication failures."""
with patch('pymssql.connect') as mock_connect:
mock_connect.side_effect = pymssql.OperationalError("Login failed for user 'test'")
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'wrong_password',
'MSSQL_DATABASE': 'testdb'
}):
resources = await app.list_resources()
assert resources == []
@pytest.mark.asyncio
async def test_database_not_found(self):
"""Test handling when database doesn't exist."""
with patch('pymssql.connect') as mock_connect:
mock_connect.side_effect = pymssql.OperationalError("Database 'nonexistent' does not exist")
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'nonexistent'
}):
resources = await app.list_resources()
assert resources == []
@pytest.mark.asyncio
async def test_network_disconnection(self):
"""Test handling of network disconnections during query."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Simulate network error during query execution
mock_cursor.execute.side_effect = pymssql.OperationalError("Network error")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM users"})
assert "Error executing query" in result[0].text
# Ensure cleanup attempted
mock_cursor.close.assert_called()
mock_conn.close.assert_called()
class TestQueryErrors:
"""Test various query execution error scenarios."""
@pytest.mark.asyncio
async def test_syntax_error(self):
"""Test handling of SQL syntax errors."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.execute.side_effect = pymssql.ProgrammingError("Incorrect syntax near 'SELCT'")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {"query": "SELCT * FROM users"})
assert "Error executing query" in result[0].text
assert len(result) == 1
@pytest.mark.asyncio
async def test_permission_denied(self):
"""Test handling of permission denied errors."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.execute.side_effect = pymssql.DatabaseError("The SELECT permission was denied")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM sensitive_table"})
assert "Error executing query" in result[0].text
@pytest.mark.asyncio
async def test_deadlock_handling(self):
"""Test handling of database deadlocks."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.execute.side_effect = pymssql.OperationalError("Transaction was deadlocked")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {
"query": "UPDATE users SET status = 'active'"
})
assert "Error executing query" in result[0].text
class TestResourceErrors:
"""Test resource access error scenarios."""
@pytest.mark.asyncio
async def test_invalid_uri_format(self):
"""Test handling of invalid resource URIs."""
from pydantic import AnyUrl
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Test invalid URI scheme
with pytest.raises(ValueError, match="Invalid URI scheme"):
await app.read_resource(AnyUrl("http://invalid/uri"))
@pytest.mark.asyncio
async def test_table_not_found(self):
"""Test handling when requested table doesn't exist."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.execute.side_effect = pymssql.ProgrammingError("Invalid object name 'nonexistent'")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
from pydantic import AnyUrl
with pytest.raises(RuntimeError, match="Database error"):
await app.read_resource(AnyUrl("mssql://nonexistent/data"))
class TestRecoveryScenarios:
"""Test recovery and resilience scenarios."""
@pytest.mark.asyncio
async def test_connection_retry_logic(self):
"""Test that connection failures don't crash the server."""
attempt_count = 0
def mock_connect(**kwargs):
nonlocal attempt_count
attempt_count += 1
if attempt_count < 3:
raise pymssql.OperationalError("Connection failed")
# Success on third attempt
mock_conn = Mock()
mock_cursor = Mock()
mock_cursor.fetchall.return_value = [('users',)]
mock_conn.cursor.return_value = mock_cursor
return mock_conn
with patch('pymssql.connect', side_effect=mock_connect):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# First two calls should fail
resources1 = await app.list_resources()
assert resources1 == []
resources2 = await app.list_resources()
assert resources2 == []
# Third call should succeed
resources3 = await app.list_resources()
assert len(resources3) == 1
@pytest.mark.asyncio
async def test_partial_result_handling(self):
"""Test handling when cursor fails mid-iteration."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Simulate cursor failing during iteration
def failing_fetchall():
raise pymssql.OperationalError("Connection lost during query")
mock_cursor.execute.return_value = None
mock_cursor.fetchall = failing_fetchall
mock_cursor.description = [('id',), ('name',)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Should handle the error gracefully
from pydantic import AnyUrl
with pytest.raises(RuntimeError):
await app.read_resource(AnyUrl("mssql://users/data"))
@pytest.mark.asyncio
async def test_long_running_query_handling(self):
"""Test handling of long-running queries."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
async def slow_execute(query):
await asyncio.sleep(0.1) # Simulate slow query
return None
mock_cursor.execute = Mock(side_effect=lambda q: None)
mock_cursor.fetchall.return_value = [(1,)]
mock_cursor.description = [('count',)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Should complete without timeout
result = await app.call_tool("execute_sql", {
"query": "SELECT COUNT(*) FROM large_table"
})
assert "1" in result[0].text
class TestMemoryAndResourceManagement:
"""Test memory and resource leak prevention."""
@pytest.mark.asyncio
async def test_cursor_cleanup_on_error(self):
"""Ensure cursors are closed even on errors."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.execute.side_effect = Exception("Unexpected error")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM users"})
# Cursor should be closed despite error
mock_cursor.close.assert_called()
mock_conn.close.assert_called()
@pytest.mark.asyncio
async def test_connection_cleanup_on_exception(self):
"""Ensure connections are closed on exceptions."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Make cursor creation fail after connection
mock_conn.cursor.side_effect = Exception("Cursor creation failed")
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
resources = await app.list_resources()
assert resources == []
# Connection should still be closed
mock_conn.close.assert_called()
```
--------------------------------------------------------------------------------
/tests/test_performance.py:
--------------------------------------------------------------------------------
```python
"""Performance and load tests for production readiness."""
import pytest
import asyncio
import time
from unittest.mock import Mock, patch
from concurrent.futures import ThreadPoolExecutor
import gc
import psutil
import os
from mssql_mcp_server.server import app
class TestPerformance:
"""Test performance characteristics under load."""
@pytest.mark.asyncio
async def test_query_response_time(self):
"""Test that queries respond within acceptable time limits."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Simulate reasonable query execution
mock_cursor.description = [('id',), ('name',)]
mock_cursor.fetchall.return_value = [(i, f'user_{i}') for i in range(100)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
start_time = time.time()
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM users"})
end_time = time.time()
# Query should complete in reasonable time (< 1 second for mock)
assert end_time - start_time < 1.0
assert len(result) == 1
assert "user_99" in result[0].text
@pytest.mark.asyncio
async def test_concurrent_query_performance(self):
"""Test performance under concurrent query load."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.description = [('count',)]
mock_cursor.fetchall.return_value = [(42,)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Run 50 concurrent queries
start_time = time.time()
tasks = [
app.call_tool("execute_sql", {"query": f"SELECT COUNT(*) FROM table_{i}"})
for i in range(50)
]
results = await asyncio.gather(*tasks)
end_time = time.time()
# All queries should complete
assert len(results) == 50
assert all("42" in r[0].text for r in results)
# Should complete in reasonable time (< 5 seconds for 50 queries)
assert end_time - start_time < 5.0
@pytest.mark.asyncio
async def test_large_result_set_performance(self):
"""Test performance with large result sets."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Create large result set (10,000 rows)
large_result = [(i, f'user_{i}', f'email_{i}@test.com', i % 100) for i in range(10000)]
mock_cursor.description = [('id',), ('name',), ('email',), ('status',)]
mock_cursor.fetchall.return_value = large_result
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
start_time = time.time()
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM large_table"})
end_time = time.time()
# Should handle large results efficiently
assert len(result) == 1
lines = result[0].text.split('\n')
assert len(lines) == 10001 # Header + 10000 rows
# Should complete in reasonable time (< 10 seconds)
assert end_time - start_time < 10.0
class TestMemoryUsage:
"""Test memory usage and leak prevention."""
@pytest.mark.asyncio
async def test_memory_usage_stability(self):
"""Test that memory usage remains stable over time."""
if not hasattr(psutil.Process(), 'memory_info'):
pytest.skip("Memory monitoring not available")
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchall.return_value = [('table1',), ('table2',)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
process = psutil.Process(os.getpid())
# Get baseline memory
gc.collect()
baseline_memory = process.memory_info().rss / 1024 / 1024 # MB
# Run many operations
for _ in range(100):
await app.list_resources()
# Check memory after operations
gc.collect()
final_memory = process.memory_info().rss / 1024 / 1024 # MB
# Memory growth should be minimal (< 50 MB)
memory_growth = final_memory - baseline_memory
assert memory_growth < 50, f"Memory grew by {memory_growth} MB"
@pytest.mark.asyncio
async def test_large_data_memory_handling(self):
"""Test memory handling with large data sets."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Create very large result
def generate_large_result():
for i in range(100000):
yield (i, f'data_{i}' * 100) # Large strings
mock_cursor.description = [('id',), ('data',)]
mock_cursor.fetchall.return_value = list(generate_large_result())
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Should handle large data without excessive memory use
result = await app.call_tool("execute_sql", {"query": "SELECT * FROM big_table"})
# Result should be created
assert len(result) == 1
# Memory should be released after operation
result = None
gc.collect()
class TestLoadHandling:
"""Test system behavior under various load conditions."""
@pytest.mark.asyncio
async def test_burst_load_handling(self):
"""Test handling of sudden burst loads."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchall.return_value = [('result',)]
mock_cursor.description = [('data',)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Simulate burst of 100 requests
start_time = time.time()
tasks = []
for _ in range(100):
tasks.append(app.call_tool("execute_sql", {"query": "SELECT 1"}))
results = await asyncio.gather(*tasks, return_exceptions=True)
end_time = time.time()
# Count successful results
successful = sum(1 for r in results if not isinstance(r, Exception))
# Most requests should succeed
assert successful >= 90 # Allow 10% failure rate
# Should complete within reasonable time
assert end_time - start_time < 30.0
@pytest.mark.asyncio
async def test_sustained_load_handling(self):
"""Test handling of sustained load over time."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.fetchall.return_value = [('ok',)]
mock_cursor.description = [('status',)]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Run continuous load for 10 seconds
start_time = time.time()
request_count = 0
error_count = 0
while time.time() - start_time < 10:
try:
result = await app.call_tool("execute_sql", {"query": "SELECT 'ok'"})
request_count += 1
assert "ok" in result[0].text
except Exception:
error_count += 1
# Small delay to prevent overwhelming
await asyncio.sleep(0.01)
# Should handle sustained load
assert request_count > 500 # At least 50 req/sec
assert error_count < request_count * 0.05 # Less than 5% errors
class TestScalability:
"""Test scalability characteristics."""
@pytest.mark.asyncio
async def test_resource_scaling(self):
"""Test handling of increasing number of resources."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
# Test with different table counts
table_counts = [10, 100, 1000]
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
for count in table_counts:
# Create table list
tables = [(f'table_{i}',) for i in range(count)]
mock_cursor.fetchall.return_value = tables
start_time = time.time()
resources = await app.list_resources()
end_time = time.time()
assert len(resources) == count
# Time should scale reasonably (not exponentially)
time_per_table = (end_time - start_time) / count
assert time_per_table < 0.01 # Less than 10ms per table
@pytest.mark.asyncio
async def test_query_complexity_scaling(self):
"""Test performance with increasingly complex queries."""
mock_conn = Mock()
mock_cursor = Mock()
mock_conn.cursor.return_value = mock_cursor
with patch('pymssql.connect', return_value=mock_conn):
with patch.dict('os.environ', {
'MSSQL_USER': 'test',
'MSSQL_PASSWORD': 'test',
'MSSQL_DATABASE': 'testdb'
}):
# Test simple to complex queries
queries = [
"SELECT 1",
"SELECT * FROM users WHERE id = 1",
"SELECT u.*, o.* FROM users u JOIN orders o ON u.id = o.user_id",
"""SELECT u.name, COUNT(o.id), SUM(o.total), AVG(o.total)
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.name
HAVING COUNT(o.id) > 5
ORDER BY SUM(o.total) DESC"""
]
mock_cursor.description = [('result',)]
mock_cursor.fetchall.return_value = [('data',)]
for query in queries:
start_time = time.time()
result = await app.call_tool("execute_sql", {"query": query})
end_time = time.time()
# All queries should complete successfully
assert len(result) == 1
# Response time should be reasonable
assert end_time - start_time < 2.0
```