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."
}