sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/app.py:55-63 (handler)

  The 'describe_table' tool is registered here and calls metadata.describe_table.

  def describe_table(
      connection_id: str,
      table_name: str,
      schema: Optional[str] = None,
      database: Optional[str] = None,
  ) -> dict:
      """Describe columns, keys, and indexes for a table."""
  
      return _run_tool(lambda: metadata.describe_table(connection_id, table_name, schema, database))

• sql_query_mcp/introspection.py:148-207 (handler)

  The 'MetadataService.describe_table' method orchestrates the retrieval of table metadata using the appropriate database adapter.

  def describe_table(
      self,
      connection_id: str,
      table_name: str,
      schema: Optional[str] = None,
      database: Optional[str] = None,
  ) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      try:
          config = self._registry.get_connection_config(connection_id)
          namespace = resolve_namespace(config, schema=schema, database=database)
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              description = adapter.describe_table(conn, namespace.value, table_name)
              if not description:
                  raise QueryExecutionError(
                      f"未找到表 {namespace.value}.{table_name}，或当前用户没有访问权限"
                  )
  
          duration_ms = _elapsed_ms(started)
          self._audit.log(
              tool="describe_table",
              connection_id=connection_id,
              success=True,
              duration_ms=duration_ms,
              row_count=len(description["columns"]),
              extra={
                  "engine": config.engine,
                  namespace.field_name: namespace.value,
                  "table_name": table_name,
              },
          )
          return {
              "connection_id": connection_id,
              "engine": config.engine,
              namespace.field_name: namespace.value,
              "table_name": table_name,
              "columns": description["columns"],
              "indexes": description["indexes"],
          }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="describe_table",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              error=sanitized,
              extra=_build_audit_extra(
                  config,
                  schema=schema,
                  database=database,
                  table_name=table_name,
              ),
          )
          raise QueryExecutionError(sanitized) from exc