schema-engram-mcp

Instructions

Implementation Reference

• schema_engram_mcp/server.py:68-94 (handler)

  The tool `db_schema_get` is registered and implemented in `server.py`. It handles retrieving a database schema snapshot by ID or the most recent one.

  @mcp.tool()
  def db_schema_get(
      workspace_id: str,
      connection_name: str,
      snapshot_id: int | None = None,
  ) -> str:
      """
      Obtiene un esquema guardado. Si `snapshot_id` es None, devuelve el más reciente
      para workspace_id + connection_name.
      """
      conn = _get_conn()
      if snapshot_id is not None:
          row = storage.get_snapshot_by_id(conn, snapshot_id)
          if row is None:
              return json.dumps({"error": "snapshot not found"}, ensure_ascii=False)
          if row["workspace_id"] != workspace_id.strip() or row["connection_name"] != connection_name.strip():
              return json.dumps(
                  {"error": "snapshot_id no coincide con workspace/connection"},
                  ensure_ascii=False,
              )
      else:
          row = storage.get_latest(
              conn, workspace_id=workspace_id.strip(), connection_name=connection_name.strip()
          )
          if row is None:
              return json.dumps({"error": "no hay instantáneas"}, ensure_ascii=False)
      return json.dumps(row, ensure_ascii=False)