mcp-snowflake-server

Instructions

Implementation Reference

• src/mcp_snowflake_server/server.py:184-226 (handler)

  The main handler function for the 'describe_table' tool. Parses the fully qualified table name (database.schema.table), executes a SQL query on Snowflake's information_schema.columns to fetch column details (name, default, nullable, type, comment), and returns YAML and optional JSON output.

  async def handle_describe_table(arguments, db, *_, exclude_json_results=False):
      if not arguments or "table_name" not in arguments:
          raise ValueError("Missing table_name argument")
  
      table_spec = arguments["table_name"]
      split_identifier = table_spec.split(".")
  
      # Parse the fully qualified table name
      if len(split_identifier) < 3:
          raise ValueError("Table name must be fully qualified as 'database.schema.table'")
  
      database_name = split_identifier[0].upper()
      schema_name = split_identifier[1].upper()
      table_name = split_identifier[2].upper()
  
      query = f"""
          SELECT column_name, column_default, is_nullable, data_type, comment 
          FROM {database_name}.information_schema.columns 
          WHERE table_schema = '{schema_name}' AND table_name = '{table_name}'
      """
      data, data_id = await db.execute_query(query)
  
      output = {
          "type": "data",
          "data_id": data_id,
          "database": database_name,
          "schema": schema_name,
          "table": table_name,
          "data": data,
      }
      yaml_output = to_yaml(output)
      json_output = to_json(output)
      results: list[ResponseType] = [types.TextContent(type="text", text=yaml_output)]
      if not exclude_json_results:
          results.append(
              types.EmbeddedResource(
                  type="resource",
                  resource=types.TextResourceContents(
                      uri=f"data://{data_id}", text=json_output, mimeType="application/json"
                  ),
              )
          )
      return results

• src/mcp_snowflake_server/server.py:420-434 (registration)

  Registers the 'describe_table' tool in the all_tools list, including its name, description, input schema (requiring 'table_name' parameter), and handler function reference.

  Tool(
      name="describe_table",
      description="Get the schema information for a specific table",
      input_schema={
          "type": "object",
          "properties": {
              "table_name": {
                  "type": "string",
                  "description": "Fully qualified table name in the format 'database.schema.table'",
              },
          },
          "required": ["table_name"],
      },
      handler=handle_describe_table,
  ),

• src/mcp_snowflake_server/server.py:423-434 (schema)

  Defines the input schema for the 'describe_table' tool, specifying an object with a required 'table_name' string parameter described as fully qualified 'database.schema.table'.

      input_schema={
          "type": "object",
          "properties": {
              "table_name": {
                  "type": "string",
                  "description": "Fully qualified table name in the format 'database.schema.table'",
              },
          },
          "required": ["table_name"],
      },
      handler=handle_describe_table,
  ),