# Directory Structure
```
├── .gitignore
├── config
│ ├── config_template.ini
│ ├── providex.ini
│ └── sqlite.ini
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── odbc_mcp
│ ├── __init__.py
│ ├── b1-odbc.py
│ ├── config.py
│ ├── odbc.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python build artifacts
2 | __pycache__/
3 | *.py[cod]
4 | *$py.class
5 | *.so
6 | .Python
7 | build/
8 | develop-eggs/
9 | dist/
10 | downloads/
11 | eggs/
12 | .eggs/
13 | lib/
14 | lib64/
15 | parts/
16 | sdist/
17 | var/
18 | wheels/
19 | *.egg-info/
20 | .installed.cfg
21 | *.egg
22 |
23 | # Virtual environment
24 | env/
25 | venv/
26 | ENV/
27 | .venv/
28 |
29 | # Environment and config files with secrets
30 | .env
31 | *.log
32 |
33 | # IDE files
34 | .idea/
35 | .vscode/
36 | *.swp
37 | *.swo
38 | .DS_Store
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # ODBC MCP Server
2 |
3 | An MCP (Model Context Protocol) server that enables LLM tools like Claude Desktop to query databases via ODBC connections. This server allows Claude and other MCP clients to access, analyze, and generate insights from database data while maintaining security and read-only safeguards.
4 |
5 | ## Features
6 |
7 | - Connect to any ODBC-compatible database
8 | - Support for multiple database connections
9 | - Flexible configuration through config files or Claude Desktop settings
10 | - Read-only safeguards to prevent data modification
11 | - Easy installation with UV package manager
12 | - Detailed error reporting and logging
13 |
14 | ## Prerequisites
15 |
16 | - Python 3.10 or higher
17 | - UV package manager
18 | - ODBC drivers for your database(s) installed on your system
19 | - For Sage 100 Advanced: ProvideX ODBC driver
20 |
21 | ## Installation
22 |
23 | ```bash
24 | git clone https://github.com/tylerstoltz/mcp-odbc.git
25 | cd mcp-odbc
26 | uv venv
27 | .venv\Scripts\activate # On Mac / Linux: source .venv/bin/activate (untested)
28 | uv pip install -e .
29 | ```
30 |
31 | ## Configuration
32 |
33 | The server can be configured through:
34 |
35 | 1. A dedicated config file
36 | 2. Environment variables
37 | 3. Claude Desktop configuration
38 |
39 | ### General Configuration Setup
40 |
41 | Create a configuration file (`.ini`) with your database connection details:
42 |
43 | ```ini
44 | [SERVER]
45 | default_connection = my_database
46 | max_rows = 1000
47 | timeout = 30
48 |
49 | [my_database]
50 | dsn = MyDatabaseDSN
51 | username = your_username
52 | password = your_password
53 | readonly = true
54 | ```
55 |
56 | ### SQLite Configuration
57 |
58 | For SQLite databases with ODBC:
59 |
60 | ```ini
61 | [SERVER]
62 | default_connection = sqlite_db
63 | max_rows = 1000
64 | timeout = 30
65 |
66 | [sqlite_db]
67 | dsn = SQLite_DSN_Name
68 | readonly = true
69 | ```
70 |
71 | ### Sage 100 ProvideX Configuration
72 |
73 | ProvideX requires special configuration for compatibility. Use this minimal configuration for best results:
74 |
75 | ```ini
76 | [SERVER]
77 | default_connection = sage100
78 | max_rows = 1000
79 | timeout = 60
80 |
81 | [sage100]
82 | dsn = YOUR_PROVIDEX_DSN
83 | username = your_username
84 | password = your_password
85 | company = YOUR_COMPANY_CODE
86 | readonly = true
87 | ```
88 |
89 | **Important notes for ProvideX:**
90 | - Use a minimal configuration - adding extra parameters may cause connection issues
91 | - Always set `readonly = true` for safety
92 | - The `company` parameter is required for Sage 100 connections
93 | - Avoid changing connection attributes after connection is established
94 |
95 | ### Claude Desktop Integration
96 |
97 | To configure the server in Claude Desktop:
98 |
99 | 1. Open or create `claude_desktop_config.json`:
100 | - Windows: `%APPDATA%\Claude\claude_desktop_config.json`
101 | - macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
102 |
103 | 2. Add MCP server configuration:
104 |
105 | ```json
106 | {
107 | "mcpServers": {
108 | "odbc": {
109 | "command": "uv",
110 | "args": [
111 | "--directory",
112 | "C:\\path\\to\\mcp-odbc",
113 | "run",
114 | "odbc-mcp-server",
115 | "--config",
116 | "C:\\path\\to\\mcp-odbc\\config\\your_config.ini"
117 | ]
118 | }
119 | }
120 | }
121 | ```
122 |
123 | ## Usage
124 |
125 | ### Starting the Server Manually
126 |
127 | ```bash
128 | # Start with default configuration
129 | odbc-mcp-server
130 |
131 | # Start with a specific config file
132 | odbc-mcp-server --config path/to/config.ini
133 | ```
134 |
135 | ### Using with Claude Desktop
136 |
137 | 1. Configure the server in Claude Desktop's config file as shown above
138 | 2. Restart Claude Desktop
139 | 3. The ODBC tools will automatically appear in the MCP tools list
140 |
141 | ### Available MCP Tools
142 |
143 | The ODBC MCP server provides these tools:
144 |
145 | 1. **list-connections**: Lists all configured database connections
146 | 2. **list-available-dsns**: Lists all available DSNs on the system
147 | 3. **test-connection**: Tests a database connection and returns information
148 | 4. **list-tables**: Lists all tables in the database
149 | 5. **get-table-schema**: Gets schema information for a table
150 | 6. **execute-query**: Executes an SQL query and returns results
151 |
152 | ## Example Queries
153 |
154 | Try these prompts in Claude Desktop after connecting the server:
155 |
156 | - "Show me all the tables in the database"
157 | - "What's the schema of the Customer table?"
158 | - "Run a query to get the first 10 customers"
159 | - "Find all orders placed in the last 30 days"
160 | - "Analyze the sales data by region and provide insights"
161 |
162 | ## Troubleshooting
163 |
164 | ### Connection Issues
165 |
166 | If you encounter connection problems:
167 |
168 | 1. Verify your ODBC drivers are installed correctly
169 | 2. Test your DSN using the ODBC Data Source Administrator
170 | 3. Check connection parameters in your config file
171 | 4. Look for detailed error messages in Claude Desktop logs
172 |
173 | ### ProvideX-Specific Issues
174 |
175 | For Sage 100/ProvideX:
176 | 1. Use minimal connection configuration (DSN, username, password, company)
177 | 2. Make sure the Company parameter is correct
178 | 3. Use the special ProvideX configuration template
179 | 4. If you encounter `Driver not capable` errors, check that autocommit is being set at connection time
180 |
181 | ### Missing Tables
182 |
183 | If tables aren't showing up:
184 |
185 | 1. Verify user permissions for the database account
186 | 2. Check if the company code is correct (for Sage 100)
187 | 3. Try using fully qualified table names (schema.table)
188 |
189 | ## License
190 |
191 | MIT License - Copyright (c) 2024
```
--------------------------------------------------------------------------------
/config/sqlite.ini:
--------------------------------------------------------------------------------
```
1 | ; SQLite ODBC Configuration
2 | ; Example configuration for connecting to a SQLite database via ODBC
3 |
4 | [SERVER]
5 | ; Default connection to use if not specified
6 | default_connection = sqlite_db
7 | ; Maximum rows to return per query (default: 1000)
8 | max_rows = 1000
9 | ; Query timeout in seconds (default: 30)
10 | timeout = 30
11 |
12 | [sqlite_db]
13 | ; DSN-based connection - replace with your SQLite DSN name
14 | dsn = SQLite_DSN_Name
15 | ; SQLite typically doesn't need username/password
16 | ; readonly = true enforces read-only mode for safety
17 | readonly = true
```
--------------------------------------------------------------------------------
/config/config_template.ini:
--------------------------------------------------------------------------------
```
1 | ; ODBC MCP Server Configuration Template
2 | ; Copy this file and rename to config.ini
3 |
4 | [SERVER]
5 | ; Default connection to use if not specified
6 | default_connection = example_dsn
7 | ; Maximum rows to return per query (default: 1000)
8 | max_rows = 1000
9 | ; Query timeout in seconds (default: 30)
10 | timeout = 30
11 |
12 | ; Example DSN-based connection
13 | [example_dsn]
14 | dsn = MyDSN
15 | username = username
16 | password = password
17 | ; Enforce read-only mode (default: true)
18 | readonly = true
19 |
20 | ; Example full connection string
21 | [example_connstr]
22 | connection_string = Driver={SQL Server};Server=myserver;Database=mydatabase;UID=username;PWD=password;
23 | readonly = true
24 |
25 | ; Example DSN-less connection
26 | [example_dsnless]
27 | driver = SQL Server
28 | server = myserver
29 | database = mydatabase
30 | username = username
31 | password = password
32 | readonly = true
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [build-system]
2 | requires = ["hatchling"]
3 | build-backend = "hatchling.build"
4 |
5 | [project]
6 | name = "odbc-mcp-server"
7 | version = "0.1.0"
8 | description = "MCP server for ODBC database connections"
9 | readme = "README.md"
10 | requires-python = ">=3.10"
11 | license = { text = "MIT" }
12 | authors = [
13 | { name = "Tyler Stoltz", email = "[email protected]" }
14 | ]
15 | classifiers = [
16 | "Programming Language :: Python :: 3",
17 | "License :: OSI Approved :: MIT License",
18 | "Operating System :: Microsoft :: Windows",
19 | ]
20 | dependencies = [
21 | "mcp>=0.1.0",
22 | "pyodbc>=4.0.34",
23 | "pydantic>=2.0.0",
24 | "configparser>=5.0.0",
25 | ]
26 |
27 | [project.scripts]
28 | odbc-mcp-server = "odbc_mcp:main"
29 |
30 | [project.optional-dependencies]
31 | dev = [
32 | "black>=23.0.0",
33 | "pytest>=7.0.0",
34 | ]
35 |
36 | [tool.hatch.build.targets.wheel]
37 | packages = ["src/odbc_mcp"]
38 |
39 | [tool.pytest.ini_options]
40 | testpaths = ["tests"]
```
--------------------------------------------------------------------------------
/config/providex.ini:
--------------------------------------------------------------------------------
```
1 | ; Sage 100 ProvideX ODBC Configuration
2 | ; Minimal configuration for best compatibility with Sage 100 Advanced
3 |
4 | [SERVER]
5 | ; Default connection to use if not specified
6 | default_connection = sage100
7 | ; Maximum rows to return per query
8 | max_rows = 1000
9 | ; Query timeout in seconds
10 | timeout = 60
11 |
12 | [sage100]
13 | ; IMPORTANT: For ProvideX compatibility, use minimal settings
14 | ; Replace with your ProvideX DSN (usually something like SOTAMAS90)
15 | dsn = YOUR_PROVIDEX_DSN
16 | ; Replace with your actual username
17 | username = YOUR_USERNAME
18 | ; Replace with your actual password
19 | password = YOUR_PASSWORD
20 | ; Company code (required for Sage 100) - typically a 3-letter code
21 | company = YOUR_COMPANY_CODE
22 | ; Always keep readonly mode for safety
23 | readonly = true
24 |
25 | ; NOTE: For ProvideX compatibility:
26 | ; - Use minimal settings as shown above
27 | ; - Avoid adding extra parameters that may cause connection issues
28 | ; - The connection uses autocommit=True automatically for ProvideX drivers
```
--------------------------------------------------------------------------------
/src/odbc_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | ODBC MCP Server package.
3 | Provides MCP tools for querying databases via ODBC.
4 | """
5 |
6 | import asyncio
7 | import argparse
8 | import sys
9 | from pathlib import Path
10 | from .server import ODBCMCPServer
11 |
12 |
13 | def main():
14 | """Main entry point for the package."""
15 | parser = argparse.ArgumentParser(description="ODBC MCP Server")
16 | parser.add_argument(
17 | "--config", "-c",
18 | help="Path to configuration file",
19 | type=str
20 | )
21 | args = parser.parse_args()
22 |
23 | try:
24 | server = ODBCMCPServer(args.config)
25 | asyncio.run(server.run())
26 | except KeyboardInterrupt:
27 | print("Server shutting down...")
28 | sys.exit(0)
29 | except Exception as e:
30 | print(f"Error: {e}", file=sys.stderr)
31 | sys.exit(1)
32 |
33 |
34 | # Expose key classes at package level
35 | from .config import ServerConfig, ODBCConnection
36 | from .odbc import ODBCHandler
37 |
38 | __all__ = ['ODBCMCPServer', 'ServerConfig', 'ODBCConnection', 'ODBCHandler', 'main']
```
--------------------------------------------------------------------------------
/src/odbc_mcp/config.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Configuration management for ODBC MCP Server.
3 | Handles loading and validating ODBC connection settings.
4 | """
5 |
6 | import os
7 | import json
8 | from pathlib import Path
9 | from typing import Dict, Optional, List, Any
10 | import configparser
11 | from pydantic import BaseModel, Field, validator
12 |
13 |
14 | class ODBCConnection(BaseModel):
15 | """ODBC connection configuration model."""
16 | name: str
17 | connection_string: Optional[str] = None
18 | dsn: Optional[str] = None
19 | username: Optional[str] = None
20 | password: Optional[str] = None
21 | driver: Optional[str] = None
22 | server: Optional[str] = None
23 | database: Optional[str] = None
24 | additional_params: Dict[str, str] = Field(default_factory=dict)
25 | readonly: bool = True # Enforce read-only mode
26 |
27 | @validator('connection_string', 'dsn', 'username', 'password', 'driver', 'server', 'database', pre=True)
28 | def empty_str_to_none(cls, v):
29 | """Convert empty strings to None."""
30 | if v == "":
31 | return None
32 | return v
33 |
34 | def get_connection_string(self) -> str:
35 | """Generate complete connection string for pyodbc."""
36 | # If a full connection string is provided, use it
37 | if self.connection_string:
38 | return self.connection_string
39 |
40 | # Otherwise build from components
41 | parts = []
42 |
43 | if self.dsn:
44 | parts.append(f"DSN={self.dsn}")
45 | if self.driver:
46 | parts.append(f"Driver={{{self.driver}}}")
47 | if self.server:
48 | parts.append(f"Server={self.server}")
49 | if self.database:
50 | parts.append(f"Database={self.database}")
51 | if self.username:
52 | parts.append(f"UID={self.username}")
53 | if self.password:
54 | parts.append(f"PWD={self.password}")
55 |
56 | # Add any additional parameters
57 | for key, value in self.additional_params.items():
58 | parts.append(f"{key}={value}")
59 |
60 | return ";".join(parts)
61 |
62 |
63 | class ServerConfig(BaseModel):
64 | """Main server configuration."""
65 | connections: Dict[str, ODBCConnection] = Field(default_factory=dict)
66 | default_connection: Optional[str] = None
67 | max_rows: int = 1000 # Default limit for query results
68 | timeout: int = 30 # Default timeout in seconds
69 |
70 | @validator('default_connection')
71 | def check_default_connection(cls, v, values):
72 | """Ensure default_connection references a valid connection."""
73 | if v is not None and v not in values.get('connections', {}):
74 | raise ValueError(f"Default connection '{v}' not found in configured connections")
75 | return v
76 |
77 |
78 | def load_from_ini(file_path: str) -> ServerConfig:
79 | """Load configuration from an INI file."""
80 | if not os.path.exists(file_path):
81 | raise FileNotFoundError(f"Configuration file not found: {file_path}")
82 |
83 | config = configparser.ConfigParser()
84 | config.read(file_path)
85 |
86 | connections = {}
87 | default_connection = None
88 | max_rows = 1000
89 | timeout = 30
90 |
91 | # Extract server config
92 | if 'SERVER' in config:
93 | server_config = config['SERVER']
94 | default_connection = server_config.get('default_connection')
95 | max_rows = server_config.getint('max_rows', 1000)
96 | timeout = server_config.getint('timeout', 30)
97 |
98 | # Extract connection configs
99 | for section in config.sections():
100 | if section == 'SERVER':
101 | continue
102 |
103 | # This is a connection section
104 | connection_config = dict(config[section])
105 |
106 | # Handle additional parameters (anything not specifically processed)
107 | additional_params = {}
108 | for key, value in connection_config.items():
109 | if key not in ['connection_string', 'dsn', 'username', 'password',
110 | 'driver', 'server', 'database', 'readonly']:
111 | additional_params[key] = value
112 |
113 | # Create the connection object
114 | readonly = connection_config.get('readonly', 'true').lower() in ['true', 'yes', '1', 'on']
115 | connection = ODBCConnection(
116 | name=section,
117 | connection_string=connection_config.get('connection_string'),
118 | dsn=connection_config.get('dsn'),
119 | username=connection_config.get('username'),
120 | password=connection_config.get('password'),
121 | driver=connection_config.get('driver'),
122 | server=connection_config.get('server'),
123 | database=connection_config.get('database'),
124 | additional_params=additional_params,
125 | readonly=readonly
126 | )
127 |
128 | connections[section] = connection
129 |
130 | return ServerConfig(
131 | connections=connections,
132 | default_connection=default_connection,
133 | max_rows=max_rows,
134 | timeout=timeout
135 | )
136 |
137 |
138 | def load_from_claude_config(claude_config_path: Optional[str] = None) -> Optional[ServerConfig]:
139 | """
140 | Load configuration from Claude Desktop's config file.
141 |
142 | Looks for a configuration section under mcpServerEnv.odbc_mcp_server
143 | """
144 | if claude_config_path is None:
145 | # Default paths for Claude Desktop config
146 | if os.name == 'nt': # Windows
147 | claude_config_path = os.path.join(os.environ.get('APPDATA', ''), 'Claude', 'claude_desktop_config.json')
148 | else: # macOS
149 | claude_config_path = os.path.expanduser('~/Library/Application Support/Claude/claude_desktop_config.json')
150 |
151 | if not os.path.exists(claude_config_path):
152 | return None
153 |
154 | try:
155 | with open(claude_config_path, 'r') as f:
156 | claude_config = json.load(f)
157 |
158 | # Check if our server config exists
159 | if 'mcpServerEnv' not in claude_config or 'odbc_mcp_server' not in claude_config['mcpServerEnv']:
160 | return None
161 |
162 | odbc_config = claude_config['mcpServerEnv']['odbc_mcp_server']
163 |
164 | # Process connections
165 | connections = {}
166 | for conn_name, conn_config in odbc_config.get('connections', {}).items():
167 | connections[conn_name] = ODBCConnection(
168 | name=conn_name,
169 | **conn_config
170 | )
171 |
172 | return ServerConfig(
173 | connections=connections,
174 | default_connection=odbc_config.get('default_connection'),
175 | max_rows=odbc_config.get('max_rows', 1000),
176 | timeout=odbc_config.get('timeout', 30)
177 | )
178 | except Exception as e:
179 | print(f"Error loading Claude config: {e}")
180 | return None
181 |
182 |
183 | def load_config(config_path: Optional[str] = None) -> ServerConfig:
184 | """
185 | Load configuration from file or Claude Desktop config.
186 |
187 | Order of precedence:
188 | 1. Specified config file path
189 | 2. ENV variable ODBC_MCP_CONFIG
190 | 3. Claude Desktop config
191 | 4. Default config path (./config/config.ini)
192 | """
193 | # Check specified path
194 | if config_path and os.path.exists(config_path):
195 | return load_from_ini(config_path)
196 |
197 | # Check environment variable
198 | env_config_path = os.environ.get('ODBC_MCP_CONFIG')
199 | if env_config_path and os.path.exists(env_config_path):
200 | return load_from_ini(env_config_path)
201 |
202 | # Try Claude Desktop config
203 | claude_config = load_from_claude_config()
204 | if claude_config:
205 | return claude_config
206 |
207 | # Try default path
208 | default_path = os.path.join(os.path.dirname(__file__), '..', '..', 'config', 'config.ini')
209 | if os.path.exists(default_path):
210 | return load_from_ini(default_path)
211 |
212 | # If no config found, raise error
213 | raise FileNotFoundError(
214 | "No configuration found. Please provide a config file or set the ODBC_MCP_CONFIG environment variable."
215 | )
```
--------------------------------------------------------------------------------
/src/odbc_mcp/server.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | MCP Server implementation for ODBC connections.
3 | Provides tools for database querying via the Model Context Protocol.
4 | """
5 |
6 | import asyncio
7 | import os
8 | import sys
9 | import json
10 | import logging
11 | from typing import Dict, List, Any, Optional
12 |
13 | from mcp.server import Server, NotificationOptions
14 | from mcp.server.models import InitializationOptions
15 | from mcp.server.stdio import stdio_server
16 | import mcp.types as types
17 |
18 | from .config import load_config, ServerConfig
19 | from .odbc import ODBCHandler
20 |
21 |
22 | # Configure logging
23 | logging.basicConfig(
24 | level=logging.INFO,
25 | format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
26 | handlers=[
27 | logging.StreamHandler(sys.stderr)
28 | ]
29 | )
30 |
31 | logger = logging.getLogger("odbc-mcp-server")
32 |
33 |
34 | class ODBCMCPServer:
35 | """
36 | MCP Server that provides tools for ODBC database connectivity.
37 | """
38 |
39 | def __init__(self, config_path: Optional[str] = None):
40 | """Initialize the server with configuration."""
41 | try:
42 | self.config = load_config(config_path)
43 | self.odbc = ODBCHandler(self.config)
44 | self.server = Server("odbc-mcp-server")
45 |
46 | # Register tool handlers
47 | self._register_tools()
48 |
49 | logger.info(f"Initialized ODBC MCP Server with {len(self.config.connections)} connections")
50 | except Exception as e:
51 | logger.error(f"Failed to initialize server: {e}")
52 | raise
53 |
54 | def _register_tools(self):
55 | """Register all MCP tools."""
56 | @self.server.list_tools()
57 | async def list_tools() -> List[types.Tool]:
58 | """List available tools for the MCP client."""
59 | return [
60 | types.Tool(
61 | name="list-connections",
62 | description="List all configured database connections",
63 | inputSchema={
64 | "type": "object",
65 | "properties": {},
66 | "required": []
67 | }
68 | ),
69 | types.Tool(
70 | name="list-available-dsns",
71 | description="List all available DSNs on the system",
72 | inputSchema={
73 | "type": "object",
74 | "properties": {},
75 | "required": []
76 | }
77 | ),
78 | types.Tool(
79 | name="test-connection",
80 | description="Test a database connection and return information",
81 | inputSchema={
82 | "type": "object",
83 | "properties": {
84 | "connection_name": {
85 | "type": "string",
86 | "description": "Name of the connection to test (optional, uses default if not specified)"
87 | }
88 | },
89 | "required": []
90 | }
91 | ),
92 | types.Tool(
93 | name="list-tables",
94 | description="List all tables in the database",
95 | inputSchema={
96 | "type": "object",
97 | "properties": {
98 | "connection_name": {
99 | "type": "string",
100 | "description": "Name of the connection to use (optional, uses default if not specified)"
101 | }
102 | },
103 | "required": []
104 | }
105 | ),
106 | types.Tool(
107 | name="get-table-schema",
108 | description="Get schema information for a table",
109 | inputSchema={
110 | "type": "object",
111 | "properties": {
112 | "table_name": {
113 | "type": "string",
114 | "description": "Name of the table to describe (required)"
115 | },
116 | "connection_name": {
117 | "type": "string",
118 | "description": "Name of the connection to use (optional, uses default if not specified)"
119 | }
120 | },
121 | "required": ["table_name"]
122 | }
123 | ),
124 | types.Tool(
125 | name="execute-query",
126 | description="Execute an SQL query and return results",
127 | inputSchema={
128 | "type": "object",
129 | "properties": {
130 | "sql": {
131 | "type": "string",
132 | "description": "SQL query to execute (required)"
133 | },
134 | "connection_name": {
135 | "type": "string",
136 | "description": "Name of the connection to use (optional, uses default if not specified)"
137 | },
138 | "max_rows": {
139 | "type": "integer",
140 | "description": "Maximum number of rows to return (optional, uses default if not specified)"
141 | }
142 | },
143 | "required": ["sql"]
144 | }
145 | )
146 | ]
147 |
148 | @self.server.call_tool()
149 | async def call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent]:
150 | """Handle tool execution requests."""
151 | arguments = arguments or {}
152 |
153 | try:
154 | if name == "list-connections":
155 | connections = self.odbc.list_connections()
156 | result = {
157 | "connections": connections,
158 | "default_connection": self.config.default_connection
159 | }
160 | return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
161 |
162 | elif name == "list-available-dsns":
163 | dsns = self.odbc.get_available_dsns()
164 | return [types.TextContent(type="text", text=json.dumps(dsns, indent=2))]
165 |
166 | elif name == "test-connection":
167 | connection_name = arguments.get("connection_name")
168 | result = self.odbc.test_connection(connection_name)
169 | return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
170 |
171 | elif name == "list-tables":
172 | connection_name = arguments.get("connection_name")
173 | tables = self.odbc.list_tables(connection_name)
174 |
175 | # Format the results for better readability
176 | result_text = "### Tables:\n\n"
177 | for table in tables:
178 | schema_prefix = f"{table['schema']}." if table['schema'] else ""
179 | result_text += f"- {schema_prefix}{table['name']}\n"
180 |
181 | return [types.TextContent(type="text", text=result_text)]
182 |
183 | elif name == "get-table-schema":
184 | table_name = arguments.get("table_name")
185 | if not table_name:
186 | raise ValueError("Table name is required")
187 |
188 | connection_name = arguments.get("connection_name")
189 | columns = self.odbc.get_table_schema(table_name, connection_name)
190 |
191 | # Format the results for better readability
192 | result_text = f"### Schema for table {table_name}:\n\n"
193 | result_text += "| Column | Type | Size | Nullable |\n"
194 | result_text += "| ------ | ---- | ---- | -------- |\n"
195 |
196 | for column in columns:
197 | result_text += f"| {column['name']} | {column['type']} | {column['size']} | {'Yes' if column['nullable'] else 'No'} |\n"
198 |
199 | return [types.TextContent(type="text", text=result_text)]
200 |
201 | elif name == "execute-query":
202 | sql = arguments.get("sql")
203 | if not sql:
204 | raise ValueError("SQL query is required")
205 |
206 | connection_name = arguments.get("connection_name")
207 | max_rows = arguments.get("max_rows")
208 |
209 | column_names, rows = self.odbc.execute_query(sql, connection_name, max_rows)
210 |
211 | # Format the results as a markdown table
212 | if not column_names:
213 | return [types.TextContent(type="text", text="Query executed successfully, but no results were returned.")]
214 |
215 | # Create the results table
216 | result_text = "### Query Results:\n\n"
217 |
218 | # Add the header row
219 | result_text += "| " + " | ".join(column_names) + " |\n"
220 |
221 | # Add the separator row
222 | result_text += "| " + " | ".join(["---"] * len(column_names)) + " |\n"
223 |
224 | # Add the data rows
225 | for row in rows:
226 | result_text += "| " + " | ".join(str(value) if value is not None else "NULL" for value in row) + " |\n"
227 |
228 | # Add the row count
229 | result_text += f"\n\n_Returned {len(rows)} rows_"
230 |
231 | # Check if we hit the row limit
232 | if max_rows and len(rows) >= max_rows:
233 | result_text += f" _(limited to {max_rows} rows)_"
234 |
235 | return [types.TextContent(type="text", text=result_text)]
236 |
237 | else:
238 | raise ValueError(f"Unknown tool: {name}")
239 |
240 | except Exception as e:
241 | logger.error(f"Error executing tool {name}: {e}")
242 | error_message = f"Error executing {name}: {str(e)}"
243 | return [types.TextContent(type="text", text=error_message)]
244 |
245 | async def run(self):
246 | """Run the MCP server."""
247 | try:
248 | initialization_options = InitializationOptions(
249 | server_name="odbc-mcp-server",
250 | server_version="0.1.0",
251 | capabilities=self.server.get_capabilities(
252 | notification_options=NotificationOptions(),
253 | experimental_capabilities={},
254 | ),
255 | )
256 |
257 | async with stdio_server() as (read_stream, write_stream):
258 | logger.info("Starting ODBC MCP Server")
259 | await self.server.run(
260 | read_stream,
261 | write_stream,
262 | initialization_options,
263 | )
264 | except Exception as e:
265 | logger.error(f"Server error: {e}")
266 | raise
267 | finally:
268 | # Clean up connections
269 | self.odbc.close_all_connections()
```
--------------------------------------------------------------------------------
/src/odbc_mcp/b1-odbc.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | ODBC connection and query management.
3 | Handles database connections and provides methods for executing queries.
4 | """
5 |
6 | import pyodbc
7 | import re
8 | from typing import List, Dict, Any, Optional, Tuple, Union
9 | from .config import ODBCConnection, ServerConfig
10 |
11 |
12 | class ODBCHandler:
13 | """Handles ODBC connections and query execution."""
14 |
15 | def __init__(self, config: ServerConfig):
16 | """Initialize with server configuration."""
17 | self.config = config
18 | self.connections = config.connections
19 | self.default_connection = config.default_connection
20 | self.max_rows = config.max_rows
21 | self.timeout = config.timeout
22 | self.active_connections: Dict[str, pyodbc.Connection] = {}
23 |
24 | def __del__(self):
25 | """Ensure all connections are closed on deletion."""
26 | self.close_all_connections()
27 |
28 | def close_all_connections(self):
29 | """Close all active database connections."""
30 | for conn_name, conn in self.active_connections.items():
31 | try:
32 | conn.close()
33 | except Exception:
34 | pass
35 | self.active_connections = {}
36 |
37 | def get_connection(self, connection_name: Optional[str] = None) -> pyodbc.Connection:
38 | """
39 | Get a database connection by name or use the default.
40 |
41 | Args:
42 | connection_name: Name of the connection to use, or None for default
43 |
44 | Returns:
45 | pyodbc.Connection: Active database connection
46 |
47 | Raises:
48 | ValueError: If connection name doesn't exist
49 | ConnectionError: If connection fails
50 | """
51 | # Use default if not specified
52 | if connection_name is None:
53 | if self.default_connection is None:
54 | if len(self.connections) == 1:
55 | # If only one connection is defined, use it
56 | connection_name = list(self.connections.keys())[0]
57 | else:
58 | raise ValueError("No default connection specified and multiple connections exist")
59 | else:
60 | connection_name = self.default_connection
61 |
62 | # Check if connection exists
63 | if connection_name not in self.connections:
64 | raise ValueError(f"Connection '{connection_name}' not found in configuration")
65 |
66 | # Return existing connection if available
67 | if connection_name in self.active_connections:
68 | try:
69 | # Test the connection with a simple query
70 | self.active_connections[connection_name].cursor().execute("SELECT 1")
71 | return self.active_connections[connection_name]
72 | except Exception:
73 | # Connection is stale, close it
74 | try:
75 | self.active_connections[connection_name].close()
76 | except Exception:
77 | pass
78 | del self.active_connections[connection_name]
79 |
80 | # Create new connection
81 | connection_config = self.connections[connection_name]
82 | conn_str = connection_config.get_connection_string()
83 |
84 | try:
85 | connection = pyodbc.connect(conn_str, timeout=self.timeout)
86 | # Set read-only if specified
87 | if connection_config.readonly:
88 | connection.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
89 | connection.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
90 | connection.setencoding(encoding='utf-8')
91 |
92 | self.active_connections[connection_name] = connection
93 | return connection
94 | except Exception as e:
95 | raise ConnectionError(f"Failed to connect to '{connection_name}': {str(e)}")
96 |
97 | def list_connections(self) -> List[str]:
98 | """List all available connection names."""
99 | return list(self.connections.keys())
100 |
101 | def get_available_dsns(self) -> List[Dict[str, str]]:
102 | """
103 | Get a list of all available DSNs on the system.
104 |
105 | Returns:
106 | List of dictionaries containing DSN information
107 | """
108 | dsns = []
109 | for dsn_info in pyodbc.dataSources().items():
110 | dsns.append({
111 | "name": dsn_info[0],
112 | "driver": dsn_info[1]
113 | })
114 | return dsns
115 |
116 | def list_tables(self, connection_name: Optional[str] = None) -> List[Dict[str, str]]:
117 | """
118 | List all tables in the database.
119 |
120 | Args:
121 | connection_name: Name of the connection to use, or None for default
122 |
123 | Returns:
124 | List of dictionaries with table information
125 | """
126 | connection = self.get_connection(connection_name)
127 | cursor = connection.cursor()
128 |
129 | tables = []
130 | try:
131 | for table_info in cursor.tables():
132 | if table_info.table_type == 'TABLE':
133 | tables.append({
134 | "catalog": table_info.table_cat or "",
135 | "schema": table_info.table_schem or "",
136 | "name": table_info.table_name,
137 | "type": table_info.table_type
138 | })
139 | return tables
140 | except Exception as e:
141 | # For some ODBC drivers that don't support table enumeration,
142 | # fallback to a SQL query if possible
143 | try:
144 | sql_tables = []
145 | cursor.execute("SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, TABLE_TYPE FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")
146 | for row in cursor.fetchall():
147 | sql_tables.append({
148 | "catalog": row[0] or "",
149 | "schema": row[1] or "",
150 | "name": row[2],
151 | "type": row[3]
152 | })
153 | return sql_tables
154 | except Exception:
155 | # If everything fails, raise the original error
156 | raise ConnectionError(f"Failed to list tables: {str(e)}")
157 |
158 | def get_table_schema(self, table_name: str, connection_name: Optional[str] = None) -> List[Dict[str, Any]]:
159 | """
160 | Get schema information for a table.
161 |
162 | Args:
163 | table_name: Name of the table
164 | connection_name: Name of the connection to use, or None for default
165 |
166 | Returns:
167 | List of dictionaries with column information
168 | """
169 | connection = self.get_connection(connection_name)
170 | cursor = connection.cursor()
171 |
172 | # Try to extract schema and table name
173 | schema_parts = table_name.split('.')
174 | if len(schema_parts) > 1:
175 | schema_name = schema_parts[0]
176 | table_name = schema_parts[1]
177 | else:
178 | schema_name = None
179 |
180 | columns = []
181 | try:
182 | # Use metadata API if available
183 | column_metadata = cursor.columns(table=table_name, schema=schema_name)
184 | for column in column_metadata:
185 | columns.append({
186 | "name": column.column_name,
187 | "type": column.type_name,
188 | "size": column.column_size,
189 | "nullable": column.nullable == 1,
190 | "position": column.ordinal_position
191 | })
192 |
193 | # If we got column info, return it
194 | if columns:
195 | return columns
196 |
197 | # Otherwise, try SQL approach
198 | raise Exception("No columns found")
199 | except Exception:
200 | # Try SQL approach for drivers that don't support metadata
201 | try:
202 | sql = f"SELECT * FROM {table_name} WHERE 1=0"
203 | cursor.execute(sql)
204 |
205 | columns = []
206 | for i, column in enumerate(cursor.description):
207 | columns.append({
208 | "name": column[0],
209 | "type": self._get_type_name(column[1]),
210 | "size": column[3],
211 | "nullable": column[6] == 1,
212 | "position": i+1
213 | })
214 | return columns
215 | except Exception as e:
216 | raise ValueError(f"Failed to get schema for table '{table_name}': {str(e)}")
217 |
218 | def _get_type_name(self, type_code: int) -> str:
219 | """Convert ODBC type code to type name."""
220 | type_map = {
221 | pyodbc.SQL_CHAR: "CHAR",
222 | pyodbc.SQL_VARCHAR: "VARCHAR",
223 | pyodbc.SQL_LONGVARCHAR: "LONGVARCHAR",
224 | pyodbc.SQL_WCHAR: "WCHAR",
225 | pyodbc.SQL_WVARCHAR: "WVARCHAR",
226 | pyodbc.SQL_WLONGVARCHAR: "WLONGVARCHAR",
227 | pyodbc.SQL_DECIMAL: "DECIMAL",
228 | pyodbc.SQL_NUMERIC: "NUMERIC",
229 | pyodbc.SQL_SMALLINT: "SMALLINT",
230 | pyodbc.SQL_INTEGER: "INTEGER",
231 | pyodbc.SQL_REAL: "REAL",
232 | pyodbc.SQL_FLOAT: "FLOAT",
233 | pyodbc.SQL_DOUBLE: "DOUBLE",
234 | pyodbc.SQL_BIT: "BIT",
235 | pyodbc.SQL_TINYINT: "TINYINT",
236 | pyodbc.SQL_BIGINT: "BIGINT",
237 | pyodbc.SQL_BINARY: "BINARY",
238 | pyodbc.SQL_VARBINARY: "VARBINARY",
239 | pyodbc.SQL_LONGVARBINARY: "LONGVARBINARY",
240 | pyodbc.SQL_TYPE_DATE: "DATE",
241 | pyodbc.SQL_TYPE_TIME: "TIME",
242 | pyodbc.SQL_TYPE_TIMESTAMP: "TIMESTAMP",
243 | pyodbc.SQL_SS_VARIANT: "SQL_VARIANT",
244 | pyodbc.SQL_SS_UDT: "UDT",
245 | pyodbc.SQL_SS_XML: "XML",
246 | pyodbc.SQL_SS_TIME2: "TIME",
247 | pyodbc.SQL_SS_TIMESTAMPOFFSET: "TIMESTAMPOFFSET",
248 | }
249 | return type_map.get(type_code, f"UNKNOWN({type_code})")
250 |
251 | def is_read_only_query(self, sql: str) -> bool:
252 | """
253 | Check if an SQL query is read-only.
254 |
255 | Args:
256 | sql: SQL query to check
257 |
258 | Returns:
259 | bool: True if the query is read-only, False otherwise
260 | """
261 | # Remove comments and normalize whitespace
262 | sql = re.sub(r'--.*?(\n|$)', ' ', sql)
263 | sql = re.sub(r'/\*.*?\*/', ' ', sql, flags=re.DOTALL)
264 | sql = ' '.join(sql.split()).strip().upper()
265 |
266 | # Check for data modification statements
267 | data_modification_patterns = [
268 | r'^\s*INSERT\s+INTO',
269 | r'^\s*UPDATE\s+',
270 | r'^\s*DELETE\s+FROM',
271 | r'^\s*DROP\s+',
272 | r'^\s*CREATE\s+',
273 | r'^\s*ALTER\s+',
274 | r'^\s*TRUNCATE\s+',
275 | r'^\s*GRANT\s+',
276 | r'^\s*REVOKE\s+',
277 | r'^\s*MERGE\s+',
278 | r'^\s*EXEC\s+',
279 | r'^\s*EXECUTE\s+',
280 | r'^\s*CALL\s+',
281 | r'^\s*SET\s+',
282 | r'^\s*USE\s+',
283 | ]
284 |
285 | for pattern in data_modification_patterns:
286 | if re.search(pattern, sql):
287 | return False
288 |
289 | # If no modification patterns are found, it's likely read-only
290 | return True
291 |
292 | def execute_query(self, sql: str, connection_name: Optional[str] = None,
293 | max_rows: Optional[int] = None) -> Tuple[List[str], List[List[Any]]]:
294 | """
295 | Execute an SQL query and return results.
296 |
297 | Args:
298 | sql: SQL query to execute
299 | connection_name: Name of the connection to use, or None for default
300 | max_rows: Maximum number of rows to return, or None for default
301 |
302 | Returns:
303 | Tuple of column names and result rows
304 | """
305 | # Check if query is read-only for connections with readonly flag
306 | connection = self.get_connection(connection_name)
307 | connection_config = self.connections[connection_name or self.default_connection]
308 |
309 | if connection_config.readonly and not self.is_read_only_query(sql):
310 | raise ValueError("Write operations are not allowed on read-only connections")
311 |
312 | # Set max rows limit
313 | if max_rows is None:
314 | max_rows = self.max_rows
315 |
316 | # Execute the query
317 | cursor = connection.cursor()
318 | cursor.execute(sql)
319 |
320 | # Get column names
321 | column_names = [column[0] for column in cursor.description] if cursor.description else []
322 |
323 | # Fetch results with row limit
324 | results = []
325 | row_count = 0
326 |
327 | for row in cursor:
328 | formatted_row = []
329 | for value in row:
330 | # Convert specific ODBC types to strings for JSON compatibility
331 | if isinstance(value, (bytearray, bytes)):
332 | formatted_row.append(str(value))
333 | else:
334 | formatted_row.append(value)
335 | results.append(formatted_row)
336 |
337 | row_count += 1
338 | if row_count >= max_rows:
339 | break
340 |
341 | return column_names, results
342 |
343 | def test_connection(self, connection_name: Optional[str] = None) -> Dict[str, Any]:
344 | """
345 | Test a database connection and return information.
346 |
347 | Args:
348 | connection_name: Name of the connection to use, or None for default
349 |
350 | Returns:
351 | Dictionary with connection status and info
352 | """
353 | try:
354 | # Get connection
355 | conn = self.get_connection(connection_name)
356 | cursor = conn.cursor()
357 |
358 | # Get database info
359 | database_info = {}
360 |
361 | try:
362 | cursor.execute("SELECT @@version")
363 | version = cursor.fetchone()
364 | if version:
365 | database_info["version"] = version[0]
366 | except Exception:
367 | # Some databases don't support @@version
368 | pass
369 |
370 | # Get connection info
371 | conn_info = {
372 | "driver_name": conn.getinfo(pyodbc.SQL_DRIVER_NAME) if hasattr(conn, 'getinfo') else "Unknown",
373 | "driver_version": conn.getinfo(pyodbc.SQL_DRIVER_VER) if hasattr(conn, 'getinfo') else "Unknown",
374 | "database_name": conn.getinfo(pyodbc.SQL_DATABASE_NAME) if hasattr(conn, 'getinfo') else "Unknown",
375 | "dbms_name": conn.getinfo(pyodbc.SQL_DBMS_NAME) if hasattr(conn, 'getinfo') else "Unknown",
376 | "dbms_version": conn.getinfo(pyodbc.SQL_DBMS_VER) if hasattr(conn, 'getinfo') else "Unknown",
377 | }
378 |
379 | return {
380 | "status": "connected",
381 | "connection_name": connection_name or self.default_connection,
382 | "connection_info": conn_info,
383 | "database_info": database_info
384 | }
385 |
386 | except Exception as e:
387 | return {
388 | "status": "error",
389 | "connection_name": connection_name or self.default_connection,
390 | "error": str(e)
391 | }
```
--------------------------------------------------------------------------------
/src/odbc_mcp/odbc.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | ODBC connection and query management.
3 | Handles database connections and provides methods for executing queries.
4 | """
5 |
6 | import pyodbc
7 | import re
8 | from typing import List, Dict, Any, Optional, Tuple, Union
9 | from .config import ODBCConnection, ServerConfig
10 |
11 |
12 | class ODBCHandler:
13 | """Handles ODBC connections and query execution."""
14 |
15 | def __init__(self, config: ServerConfig):
16 | """Initialize with server configuration."""
17 | self.config = config
18 | self.connections = config.connections
19 | self.default_connection = config.default_connection
20 | self.max_rows = config.max_rows
21 | self.timeout = config.timeout
22 | self.active_connections: Dict[str, pyodbc.Connection] = {}
23 |
24 | def __del__(self):
25 | """Ensure all connections are closed on deletion."""
26 | self.close_all_connections()
27 |
28 | def close_all_connections(self):
29 | """Close all active database connections."""
30 | for conn_name, conn in self.active_connections.items():
31 | try:
32 | conn.close()
33 | except Exception:
34 | pass
35 | self.active_connections = {}
36 |
37 | def get_connection(self, connection_name: Optional[str] = None) -> pyodbc.Connection:
38 | """
39 | Get a database connection by name or use the default.
40 |
41 | Args:
42 | connection_name: Name of the connection to use, or None for default
43 |
44 | Returns:
45 | pyodbc.Connection: Active database connection
46 |
47 | Raises:
48 | ValueError: If connection name doesn't exist
49 | ConnectionError: If connection fails
50 | """
51 | # Use default if not specified
52 | if connection_name is None:
53 | if self.default_connection is None:
54 | if len(self.connections) == 1:
55 | # If only one connection is defined, use it
56 | connection_name = list(self.connections.keys())[0]
57 | else:
58 | raise ValueError("No default connection specified and multiple connections exist")
59 | else:
60 | connection_name = self.default_connection
61 |
62 | # Check if connection exists
63 | if connection_name not in self.connections:
64 | raise ValueError(f"Connection '{connection_name}' not found in configuration")
65 |
66 | # Return existing connection if available
67 | if connection_name in self.active_connections:
68 | try:
69 | # Test the connection with a simple query
70 | self.active_connections[connection_name].cursor().execute("SELECT 1")
71 | return self.active_connections[connection_name]
72 | except Exception:
73 | # Connection is stale, close it
74 | try:
75 | self.active_connections[connection_name].close()
76 | except Exception:
77 | pass
78 | del self.active_connections[connection_name]
79 |
80 | # Create new connection
81 | connection_config = self.connections[connection_name]
82 | conn_str = connection_config.get_connection_string()
83 |
84 | try:
85 | # Detect if this is ProvideX or has ProvideX in the connection string
86 | is_providex = "PROVIDEX" in conn_str.upper() or connection_name.upper() == "SAGE100"
87 |
88 | # Special handling for ProvideX
89 | if is_providex:
90 | # For ProvideX, explicitly set autocommit at connection time
91 | connection = pyodbc.connect(conn_str, timeout=self.timeout, autocommit=True)
92 | else:
93 | # For other drivers, use the standard connection approach
94 | connection = pyodbc.connect(conn_str, timeout=self.timeout)
95 |
96 | # Set encoding options
97 | connection.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
98 | connection.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
99 | connection.setencoding(encoding='utf-8')
100 |
101 | self.active_connections[connection_name] = connection
102 | return connection
103 | except Exception as e:
104 | raise ConnectionError(f"Failed to connect to '{connection_name}': {str(e)}")
105 |
106 | def list_connections(self) -> List[str]:
107 | """List all available connection names."""
108 | return list(self.connections.keys())
109 |
110 | def get_available_dsns(self) -> List[Dict[str, str]]:
111 | """
112 | Get a list of all available DSNs on the system.
113 |
114 | Returns:
115 | List of dictionaries containing DSN information
116 | """
117 | dsns = []
118 | for dsn_info in pyodbc.dataSources().items():
119 | dsns.append({
120 | "name": dsn_info[0],
121 | "driver": dsn_info[1]
122 | })
123 | return dsns
124 |
125 | def list_tables(self, connection_name: Optional[str] = None) -> List[Dict[str, str]]:
126 | """
127 | List all tables in the database.
128 |
129 | Args:
130 | connection_name: Name of the connection to use, or None for default
131 |
132 | Returns:
133 | List of dictionaries with table information
134 | """
135 | connection = self.get_connection(connection_name)
136 | cursor = connection.cursor()
137 |
138 | tables = []
139 | try:
140 | for table_info in cursor.tables():
141 | if table_info.table_type == 'TABLE':
142 | tables.append({
143 | "catalog": table_info.table_cat or "",
144 | "schema": table_info.table_schem or "",
145 | "name": table_info.table_name,
146 | "type": table_info.table_type
147 | })
148 | return tables
149 | except Exception as e:
150 | # For some ODBC drivers that don't support table enumeration,
151 | # fallback to a SQL query if possible
152 | try:
153 | sql_tables = []
154 | cursor.execute("SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, TABLE_TYPE FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")
155 | for row in cursor.fetchall():
156 | sql_tables.append({
157 | "catalog": row[0] or "",
158 | "schema": row[1] or "",
159 | "name": row[2],
160 | "type": row[3]
161 | })
162 | return sql_tables
163 | except Exception:
164 | # If everything fails, raise the original error
165 | raise ConnectionError(f"Failed to list tables: {str(e)}")
166 |
167 | def get_table_schema(self, table_name: str, connection_name: Optional[str] = None) -> List[Dict[str, Any]]:
168 | """
169 | Get schema information for a table.
170 |
171 | Args:
172 | table_name: Name of the table
173 | connection_name: Name of the connection to use, or None for default
174 |
175 | Returns:
176 | List of dictionaries with column information
177 | """
178 | connection = self.get_connection(connection_name)
179 | cursor = connection.cursor()
180 |
181 | # Try to extract schema and table name
182 | schema_parts = table_name.split('.')
183 | if len(schema_parts) > 1:
184 | schema_name = schema_parts[0]
185 | table_name = schema_parts[1]
186 | else:
187 | schema_name = None
188 |
189 | columns = []
190 | try:
191 | # Use metadata API if available
192 | column_metadata = cursor.columns(table=table_name, schema=schema_name)
193 | for column in column_metadata:
194 | columns.append({
195 | "name": column.column_name,
196 | "type": column.type_name,
197 | "size": column.column_size,
198 | "nullable": column.nullable == 1,
199 | "position": column.ordinal_position
200 | })
201 |
202 | # If we got column info, return it
203 | if columns:
204 | return columns
205 |
206 | # Otherwise, try SQL approach
207 | raise Exception("No columns found")
208 | except Exception:
209 | # Try SQL approach for drivers that don't support metadata
210 | try:
211 | sql = f"SELECT * FROM {table_name} WHERE 1=0"
212 | cursor.execute(sql)
213 |
214 | columns = []
215 | for i, column in enumerate(cursor.description):
216 | columns.append({
217 | "name": column[0],
218 | "type": self._get_type_name(column[1]),
219 | "size": column[3],
220 | "nullable": column[6] == 1,
221 | "position": i+1
222 | })
223 | return columns
224 | except Exception as e:
225 | raise ValueError(f"Failed to get schema for table '{table_name}': {str(e)}")
226 |
227 | def _get_type_name(self, type_code: int) -> str:
228 | """Convert ODBC type code to type name."""
229 | type_map = {
230 | pyodbc.SQL_CHAR: "CHAR",
231 | pyodbc.SQL_VARCHAR: "VARCHAR",
232 | pyodbc.SQL_LONGVARCHAR: "LONGVARCHAR",
233 | pyodbc.SQL_WCHAR: "WCHAR",
234 | pyodbc.SQL_WVARCHAR: "WVARCHAR",
235 | pyodbc.SQL_WLONGVARCHAR: "WLONGVARCHAR",
236 | pyodbc.SQL_DECIMAL: "DECIMAL",
237 | pyodbc.SQL_NUMERIC: "NUMERIC",
238 | pyodbc.SQL_SMALLINT: "SMALLINT",
239 | pyodbc.SQL_INTEGER: "INTEGER",
240 | pyodbc.SQL_REAL: "REAL",
241 | pyodbc.SQL_FLOAT: "FLOAT",
242 | pyodbc.SQL_DOUBLE: "DOUBLE",
243 | pyodbc.SQL_BIT: "BIT",
244 | pyodbc.SQL_TINYINT: "TINYINT",
245 | pyodbc.SQL_BIGINT: "BIGINT",
246 | pyodbc.SQL_BINARY: "BINARY",
247 | pyodbc.SQL_VARBINARY: "VARBINARY",
248 | pyodbc.SQL_LONGVARBINARY: "LONGVARBINARY",
249 | pyodbc.SQL_TYPE_DATE: "DATE",
250 | pyodbc.SQL_TYPE_TIME: "TIME",
251 | pyodbc.SQL_TYPE_TIMESTAMP: "TIMESTAMP",
252 | pyodbc.SQL_SS_VARIANT: "SQL_VARIANT",
253 | pyodbc.SQL_SS_UDT: "UDT",
254 | pyodbc.SQL_SS_XML: "XML",
255 | pyodbc.SQL_SS_TIME2: "TIME",
256 | pyodbc.SQL_SS_TIMESTAMPOFFSET: "TIMESTAMPOFFSET",
257 | }
258 | return type_map.get(type_code, f"UNKNOWN({type_code})")
259 |
260 | def is_read_only_query(self, sql: str) -> bool:
261 | """
262 | Check if an SQL query is read-only.
263 |
264 | Args:
265 | sql: SQL query to check
266 |
267 | Returns:
268 | bool: True if the query is read-only, False otherwise
269 | """
270 | # Remove comments and normalize whitespace
271 | sql = re.sub(r'--.*?(\n|$)', ' ', sql)
272 | sql = re.sub(r'/\*.*?\*/', ' ', sql, flags=re.DOTALL)
273 | sql = ' '.join(sql.split()).strip().upper()
274 |
275 | # Check for data modification statements
276 | data_modification_patterns = [
277 | r'^\s*INSERT\s+INTO',
278 | r'^\s*UPDATE\s+',
279 | r'^\s*DELETE\s+FROM',
280 | r'^\s*DROP\s+',
281 | r'^\s*CREATE\s+',
282 | r'^\s*ALTER\s+',
283 | r'^\s*TRUNCATE\s+',
284 | r'^\s*GRANT\s+',
285 | r'^\s*REVOKE\s+',
286 | r'^\s*MERGE\s+',
287 | r'^\s*EXEC\s+',
288 | r'^\s*EXECUTE\s+',
289 | r'^\s*CALL\s+',
290 | r'^\s*SET\s+',
291 | r'^\s*USE\s+',
292 | ]
293 |
294 | for pattern in data_modification_patterns:
295 | if re.search(pattern, sql):
296 | return False
297 |
298 | # If no modification patterns are found, it's likely read-only
299 | return True
300 |
301 | def execute_query(self, sql: str, connection_name: Optional[str] = None,
302 | max_rows: Optional[int] = None) -> Tuple[List[str], List[List[Any]]]:
303 | """
304 | Execute an SQL query and return results.
305 |
306 | Args:
307 | sql: SQL query to execute
308 | connection_name: Name of the connection to use, or None for default
309 | max_rows: Maximum number of rows to return, or None for default
310 |
311 | Returns:
312 | Tuple of column names and result rows
313 | """
314 | # Check if query is read-only for connections with readonly flag
315 | connection = self.get_connection(connection_name)
316 | connection_config = self.connections[connection_name or self.default_connection]
317 |
318 | if connection_config.readonly and not self.is_read_only_query(sql):
319 | raise ValueError("Write operations are not allowed on read-only connections")
320 |
321 | # Set max rows limit
322 | if max_rows is None:
323 | max_rows = self.max_rows
324 |
325 | # Execute the query
326 | cursor = connection.cursor()
327 | cursor.execute(sql)
328 |
329 | # Get column names
330 | column_names = [column[0] for column in cursor.description] if cursor.description else []
331 |
332 | # Fetch results with row limit
333 | results = []
334 | row_count = 0
335 |
336 | for row in cursor:
337 | formatted_row = []
338 | for value in row:
339 | # Convert specific ODBC types to strings for JSON compatibility
340 | if isinstance(value, (bytearray, bytes)):
341 | formatted_row.append(str(value))
342 | else:
343 | formatted_row.append(value)
344 | results.append(formatted_row)
345 |
346 | row_count += 1
347 | if row_count >= max_rows:
348 | break
349 |
350 | return column_names, results
351 |
352 | def test_connection(self, connection_name: Optional[str] = None) -> Dict[str, Any]:
353 | """
354 | Test a database connection and return information.
355 |
356 | Args:
357 | connection_name: Name of the connection to use, or None for default
358 |
359 | Returns:
360 | Dictionary with connection status and info
361 | """
362 | try:
363 | # Get connection
364 | conn = self.get_connection(connection_name)
365 | cursor = conn.cursor()
366 |
367 | # Get database info
368 | database_info = {}
369 |
370 | try:
371 | cursor.execute("SELECT @@version")
372 | version = cursor.fetchone()
373 | if version:
374 | database_info["version"] = version[0]
375 | except Exception:
376 | # Some databases don't support @@version
377 | pass
378 |
379 | # Get connection info
380 | conn_info = {
381 | "driver_name": conn.getinfo(pyodbc.SQL_DRIVER_NAME) if hasattr(conn, 'getinfo') else "Unknown",
382 | "driver_version": conn.getinfo(pyodbc.SQL_DRIVER_VER) if hasattr(conn, 'getinfo') else "Unknown",
383 | "database_name": conn.getinfo(pyodbc.SQL_DATABASE_NAME) if hasattr(conn, 'getinfo') else "Unknown",
384 | "dbms_name": conn.getinfo(pyodbc.SQL_DBMS_NAME) if hasattr(conn, 'getinfo') else "Unknown",
385 | "dbms_version": conn.getinfo(pyodbc.SQL_DBMS_VER) if hasattr(conn, 'getinfo') else "Unknown",
386 | }
387 |
388 | return {
389 | "status": "connected",
390 | "connection_name": connection_name or self.default_connection,
391 | "connection_info": conn_info,
392 | "database_info": database_info
393 | }
394 |
395 | except Exception as e:
396 | return {
397 | "status": "error",
398 | "connection_name": connection_name or self.default_connection,
399 | "error": str(e)
400 | }
```