Skip to main content
Glama

Trino MCP Server

by stinkgen
""" MCP resources for interacting with Trino. """ from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple from mcp.server.fastmcp import Context, FastMCP from trino_mcp.trino_client import TrinoClient def register_trino_resources(mcp: FastMCP, client: TrinoClient) -> None: """ Register Trino resources with the MCP server. Args: mcp: The MCP server instance. client: The Trino client instance. """ @mcp.resource("trino://catalog") def list_catalogs() -> List[Dict[str, Any]]: """ List all available Trino catalogs. """ return client.get_catalogs() @mcp.resource("trino://catalog/{catalog}") def get_catalog(catalog: str) -> Dict[str, Any]: """ Get information about a specific Trino catalog. """ # For now, just return basic info - could be enhanced later return {"name": catalog} @mcp.resource("trino://catalog/{catalog}/schemas") def list_schemas(catalog: str) -> List[Dict[str, Any]]: """ List all schemas in a Trino catalog. """ return client.get_schemas(catalog) @mcp.resource("trino://catalog/{catalog}/schema/{schema}") def get_schema(catalog: str, schema: str) -> Dict[str, Any]: """ Get information about a specific Trino schema. """ return {"name": schema, "catalog": catalog} @mcp.resource("trino://catalog/{catalog}/schema/{schema}/tables") def list_tables(catalog: str, schema: str) -> List[Dict[str, Any]]: """ List all tables in a Trino schema. """ return client.get_tables(catalog, schema) @mcp.resource("trino://catalog/{catalog}/schema/{schema}/table/{table}") def get_table(catalog: str, schema: str, table: str) -> Dict[str, Any]: """ Get information about a specific Trino table. """ return client.get_table_details(catalog, schema, table) @mcp.resource("trino://catalog/{catalog}/schema/{schema}/table/{table}/columns") def list_columns(catalog: str, schema: str, table: str) -> List[Dict[str, Any]]: """ List all columns in a Trino table. """ return client.get_columns(catalog, schema, table) @mcp.resource("trino://catalog/{catalog}/schema/{schema}/table/{table}/column/{column}") def get_column(catalog: str, schema: str, table: str, column: str) -> Dict[str, Any]: """ Get information about a specific Trino column. """ columns = client.get_columns(catalog, schema, table) for col in columns: if col["name"] == column: return col # If column not found, return a basic structure return { "name": column, "catalog": catalog, "schema": schema, "table": table, "error": "Column not found" } @mcp.resource("trino://query/{query_id}") def get_query_result(query_id: str) -> Dict[str, Any]: """ Get the result of a specific Trino query by its ID. """ # This is a placeholder, as we don't store query results by ID in this basic implementation # In a real implementation, you would look up the query results from a cache or storage return { "query_id": query_id, "error": "Query results not available. This resource is for demonstration purposes only." }

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/stinkgen/trino_mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server