podbc_execute_query
Execute SQL queries and retrieve results in JSONL format. Configure max rows, parameters, and connection URL for precise data extraction via SQLAlchemy connectivity.
Instructions
Execute a SQL query and return results in JSONL format.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| max_rows | No | ||
| params | No | ||
| query | Yes | ||
| url | No |
Implementation Reference
- mcp_sqlalchemy_server/server.py:287-324 (handler)The handler function that executes the SQL query on a pyodbc connection, processes results into a list of dicts with truncated long values, and returns them as a JSONL string.
def podbc_execute_query(query: str, max_rows: int = 100, params: Optional[Dict[str, Any]] = None, user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str: """ Execute a SQL query and return results in JSONL format. Args: query (str): The SQL query to execute. max_rows (int): Maximum number of rows to return. Default is 100. params (Optional[Dict[str, Any]]): Optional dictionary of parameters to pass to the query. user (Optional[str]=None): Optional username. password (Optional[str]=None): Optional password. dsn (Optional[str]=None): Optional dsn name. Returns: str: Results in JSONL format. """ try: with get_connection(True, user, password, dsn) as conn: cursor = conn.cursor() rs = cursor.execute(query) if params is None else cursor.execute(query, params) columns = [column[0] for column in rs.description] results = [] for row in rs: rs_dict = dict(zip(columns, row)) truncated_row = {key: (str(value)[:MAX_LONG_DATA] if value is not None else None) for key, value in rs_dict.items()} results.append(truncated_row) if len(results) >= max_rows: break # Convert the results to JSONL format jsonl_results = "\n".join(json.dumps(row) for row in results) # Return the JSONL formatted results return jsonl_results except pyodbc.Error as e: logging.error(f"Error executing query: {e}") raise - mcp_sqlalchemy_server/server.py:283-286 (registration)MCP decorator that registers the podbc_execute_query function as a tool with the specified name and description.
@mcp.tool( name="podbc_execute_query", description="Execute a SQL query and return results in JSONL format." ) - mcp_sqlalchemy_server/__init__.py:5-11 (registration)Imports the tool handler from server.py into the package namespace, exposing it for MCP registration.
from .server import ( podbc_get_schemas, podbc_get_tables, podbc_describe_table, podbc_filter_table_names, podbc_execute_query, podbc_execute_query_md,