podbc_execute_query_md
Run SQL queries and retrieve results formatted as Markdown tables using SQLAlchemy connectivity, enabling easy integration with any DBMS.
Instructions
Execute a SQL query and return results in Markdown table 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:326-373 (handler)The @mcp.tool decorator registers the tool, and the function podbc_execute_query_md implements the logic: connects to DB via pyodbc, executes the query, truncates long values, formats results as Markdown table, limits to max_rows.
@mcp.tool( name="podbc_execute_query_md", description="Execute a SQL query and return results in Markdown table format." ) def podbc_execute_query_md(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 Markdown table 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 Markdown table 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 # Create the Markdown table header md_table = "| " + " | ".join(columns) + " |\n" md_table += "| " + " | ".join(["---"] * len(columns)) + " |\n" # Add rows to the Markdown table for row in results: md_table += "| " + " | ".join(str(row[col]) for col in columns) + " |\n" # Return the Markdown formatted results return md_table except pyodbc.Error as e: logging.error(f"Error executing query: {e}") raise - mcp_sqlalchemy_server/server.py:327-327 (registration)The tool is registered with FastMCP using the @mcp.tool decorator specifying the name.
name="podbc_execute_query_md", - Input schema defined by function parameters with type hints: query (str, required), max_rows (int=100), params (Optional[Dict]), user/password/dsn optional.
def podbc_execute_query_md(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: