Labsco
amornpan logo

Python MSSQL MCP Server

โ˜… 27

from amornpan

A Python MCP server for Microsoft SQL Server, enabling schema inspection and SQL query execution.

๐Ÿ”ฅ๐Ÿ”ฅ๐Ÿ”ฅโœ“ VerifiedAccount requiredNeeds API keys

Python MSSQL MCP Server

Version Python MCP FastAPI License

A Model Context Protocol server implementation in Python that provides access to Microsoft SQL Server databases. This server enables Language Models to inspect table schemas and execute SQL queries through a standardized interface.

Features

Core Functionality

  • Asynchronous operation using Python's asyncio
  • Environment-based configuration using python-dotenv
  • Comprehensive logging system
  • Connection pooling and management via pyodbc
  • Error handling and recovery
  • FastAPI integration for API endpoints
  • Pydantic models for data validation
  • MSSQL connection handling with ODBC Driver

Screenshots

MCP MSSQL Server Demo

The screenshot above demonstrates the server being used with Claude to analyze and visualize SQL data.

Project Structure

PY-MCP-MSSQL/
โ”œโ”€โ”€ src/
โ”‚   โ””โ”€โ”€ mssql/
โ”‚       โ”œโ”€โ”€ __init__.py
โ”‚       โ””โ”€โ”€ server.py
โ”œโ”€โ”€ tests/
โ”‚   โ”œโ”€โ”€ __init__.py
โ”‚   โ”œโ”€โ”€ test_mssql.py
โ”‚   โ””โ”€โ”€ test_packages.py
โ”œโ”€โ”€ .env
โ”œโ”€โ”€ .env.example
โ”œโ”€โ”€ .gitignore
โ”œโ”€โ”€ README.md
โ””โ”€โ”€ requirements.txt

Directory Structure Explanation

  • src/mssql/ - Main source code directory
    • __init__.py - Package initialization
    • server.py - Main server implementation
  • tests/ - Test files directory
    • __init__.py - Test package initialization
    • test_mssql.py - MSSQL functionality tests
    • test_packages.py - Package dependency tests
  • .env - Environment configuration file (not in git)
  • .env.example - Example environment configuration
  • .gitignore - Git ignore rules
  • README.md - Project documentation
  • requirements.txt - Project dependencies

API Implementation Details

Resource Listing

@app.list_resources()
async def list_resources() -> list[Resource]
  • Lists all available tables in the database
  • Returns table names with URIs in the format mssql://<table_name>/data
  • Includes table descriptions and MIME types

Resource Reading

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str
  • Reads data from specified table
  • Accepts URIs in the format mssql://<table_name>/data
  • Returns first 100 rows in CSV format
  • Includes column headers

SQL Execution

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]
  • Executes SQL queries
  • Supports both SELECT and modification queries
  • Returns results in CSV format for SELECT queries
  • Returns affected row count for modification queries

Error Handling

The server implements comprehensive error handling for:

  • Database connection failures
  • Invalid SQL queries
  • Resource access errors
  • URI validation
  • Tool execution errors

All errors are logged and returned with appropriate error messages.

Security Features

  • Environment variable based configuration
  • Connection string security
  • Result set size limits
  • Input validation through Pydantic
  • Proper SQL query handling

Contact Information

Amornpan Phornchaicharoen

Email LinkedIn HuggingFace GitHub

Feel free to reach out to me if you have any questions about this project or would like to collaborate!


Made with โค๏ธ by Amornpan Phornchaicharoen

Author

Amornpan Phornchaicharoen