概述

这些示例展示了在受支持的Quicknode 产品中，SDK 的结构。这些示例假设您已经将SDK 初始化为 qn 来自 快速入门.

Admin API

使用Admin API 客户端来查看和管理账户资源，例如端点、团队、使用情况、计费、安全以及速率限制。

Node.js

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

const 端点 = await qn.admin.获取端点({
  limit: 20,
  sortBy: "created_at",
  排序方向: "desc",
});

用于 (const endpoint 的 端点。数据) {
  控制台.log(endpoint.id, endpoint.name, endpoint.状态);
}

Python

import asyncio
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    端点 = await qn.admin.get_endpoints(
        limit=20,
        按...排序="created_at",
        排序方向="desc",
    )

    用于 endpoint 在 端点。数据：
        打印(endpoint.id, endpoint.name, endpoint.状态)

asyncio.run(main()))

Rust

use quicknode_sdk::{admin::GetEndpointsRequest, QuicknodeSdk};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    let 参数 = GetEndpointsRequest {
        limit: 部分(20)，
        按...排序: 部分("created_at".to_string()),
        排序方向: 部分("desc".to_string()),
        ..默认::默认()
    };
    let 端点 = qn.admin.get_endpoints(&参数).await?;

    用于 endpoint 在 端点。数据 {
        println!("{} {} {}", endpoint.id, endpoint.name, endpoint.status);
    }

    好的(())
}

Ruby

require "json"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

端点 = qn.admin.get_endpoints(
    limit: 20,
    按...排序: "created_at",
    排序方向: "desc",
)

端点["数据"].每个 do |endpoint|
  puts "#{endpoint["id"]} #{endpoint["name"]} #{endpoint["状态"]}"
结束

有关完整的 API 接口，请参阅Admin API 文档。

RPC 和工具访问

您可以使用 RPC 客户端向该账户的 Tooling Access 接口endpoint 发送 JSON-RPC 请求； qn.rpc 为每次调用生成并刷新会话 JWT。请先通过以下方式启用工具访问： qn.admin 控制平面方法（需要管理员角色）。通过 网络 将呼叫转接到特定网络。

Node.js

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

// Enable Tooling Access once (idempotent; requires an admin role).
const 状态 = await qn.admin.工具访问状态();
如果 (!状态.enabled) {
  等待 qn.管理员.启用工具访问();
}

// params defaults to []; pass an array (positional) or object (named).
const blockNumber = await qn.rpc.调用("eth_blockNumber");
const 余额 = await qn.rpc.调用("eth_getBalance", ["0xabc...", "latest"]);
控制台.log(blockNumber, 余额);

// Route a call to a specific network by its multichain key.
const slot = await qn.rpc.调用("getSlot", [], "solana-mainnet");
控制台.log(插槽);

Python

import asyncio
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    # 启用工具访问权限（仅需操作一次；该操作具有幂等性；需要管理员角色）。
    状态 = 等待 qn.admin.工具访问状态()
    如果 不 状态。启用:
        等待 qn.管理员.enable_tooling_access()

    block_number = 等待 qn.rpc.调用("eth_blockNumber")
    平衡 = 等待 qn.rpc.调用("eth_getBalance", ["0xabc...", "latest"])
    打印(块号, 余额)

    # 根据多链密钥将呼叫路由到特定网络。
    插槽 = await qn.rpc.调用("getSlot", network="solana-mainnet")
    打印(插槽)

asyncio.run(main()))

Rust

使用 quicknode_sdk::QuicknodeSdk;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    // Enable Tooling Access once (idempotent; requires an admin role).
    let 状态 = qn.admin.工具访问状态().await?;
    如果 ！状态.已启用 {
        qn.管理员.enable_tooling_access().await?;
    }

    // call(method, params, network, endpoint_url). params is Option; None defaults to [].
    let 区块编号 = qn.rpc.call("eth_blockNumber", None, None, 无).await?;
    let 余额 = qn
        .rpc
        .call(
            "eth_getBalance",
            一些(serde_json::json!(["0xabc...", "latest"])),
            无,
            无,
        )
        .await？;
    println!("{:?} {:?}", block_number, balance);

    好的(())
}

Ruby

require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

# 启用工具访问权限（仅需操作一次；该操作具有幂等性；需要管理员角色）。
状态 = qn.管理员.工具访问状态
qn.管理员.enable_tooling_access 除非 状态["启用"]

block_number = qn.rpc.call(方法: "eth_blockNumber")
平衡 = qn.rpc.call(方法: "eth_getBalance", 参数: ["0xabc...", "latest"])
puts block_number
实现平衡

# 根据多链密钥将呼叫路由到特定网络。
插槽 = qn.rpc.call(方法: "getSlot", network: "solana-mainnet")
设置插槽

Streams

使用Streams 客户端来列出、查看、创建、更新、暂停、激活、删除和测试Streams 。下面的示例列出了Streams ，统计了已启用的Streams ，并模拟执行了一个过滤器操作，而不会更改账户状态。

Node.js

import { QuicknodeSdk, StreamDataset } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

const streams = await qn.streams.listStreams({ limit: 10 });
console.log(`Streams: ${streams.pageInfo.total}`);

const 已启用 = await qn.streams.getEnabledCount();
console.log(`Enabled Streams: ${enabled.total}`);

const filterFunction = Buffer.from(`
function main(data) {
  返回数据；
}
`).toString("base64");

const 结果 = await qn.streams.testFilter({
  网络: “ethereum-mainnet ”,
  数据集: StreamDataset.Block,
  block: "17811625",
  filterFunction,
});

控制台.log(结果.result);

Python

import asyncio
import base64
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    streams = await qn.streams.list_streams(limit=10)
    print(f"Streams: {streams.page_info.total}")

    已启用 = await qn.streams.get_enabled_count()
    print(f"Enabled Streams: {enabled.total}")

    filter_function = base64.b64encode(
        b"function main(data) { return data; }"
    ).解码()

    结果 = await qn.streams.test_filter(
        网络="ethereum-mainnet",
        数据集="block",
        块="17811625",
        filter_function=filter_function,
    )

    打印(结果.result)

asyncio.run(main()))

Rust

use quicknode_sdk::{
    streams::{ListStreamsParams, StreamDataset, TestFilterParams},
    QuicknodeSdk,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    let streams = qn
        .streams
        .list_streams(&ListStreamsParams {
            limit: 部分(10)，
            偏移量: 部分(0),
            ..默认::默认()
        })
        .await？;
    println!("Streams: {}", streams.page_info.total);

    let 已启用 = qn.streams.get_enabled_count(None).await?;
    println!("Enabled Streams: {}", enabled.total);

    let 结果 = qn
        .streams
        .test_filter(&TestFilterParams {
            网络: “ethereum-mainnet ”.to_string(),
            数据集: 流数据集::Block,
            block: "17811625".to_string(),
            filter_function: 一些(
                "ZnVuY3Rpb24gbWFpbihkYXRhKSB7IHJldHVybiBkYXRhOyB9".to_string(),
            ),
            filter_language: 无,
            地址簿配置: None,
        })
        .await？;

    println!("{}", result.result);
    好的(())
}

Ruby

require "base64"
require "json"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

streams = qn.streams.list_streams({ limit: 10, offset: 0 })
puts "Streams: #{streams["pageInfo"]["total"]}"

已启用 = qn.streams.get_enabled_count({})
puts "Enabled Streams: #{enabled["total"]}"

filter_function = Base64.strict_encode64("function main(data) { return data; }")

结果 = qn.streams.test_filter(
    网络: “ethereum-mainnet ”,
    数据集: "block",
    block: "17811625",
    filter_function: filter_function,
)

将结果放入["结果"]

有关目标选项、支持的数据集以及请求详情，请参阅Streams REST API 文档。

Webhooks

使用Webhooks 客户端可列出webhooks 、查看Webhook、根据模板创建Webhook、暂停或激活推送，以及获取已启用的数量。下面的示例仅供查阅。

Node.js

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

const webhooks = await qn.webhooks.listWebhooks({
  limit: 10,
  偏移量: 0,
});

用于 (const webhook of webhooks.data) {
  控制台.log(webhook.id, webhook.name, webhook.状态);
}

const 已启用 = await qn.webhooks.getEnabledCount();
console.log(`Enabled webhooks: ${enabled.total}`);

Python

import asyncio
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    webhooks = await qn.webhooks.list_webhooks(limit=10, 偏移量=0)
    用于 webhook 在 webhooks.data:
        打印(webhook.id, webhook.name, webhook.状态)

    已启用 = await qn.webhooks.get_enabled_count()
    print(f"Enabled webhooks: {enabled.total}")

asyncio.run(main()))

Rust

use quicknode_sdk::{webhooks::GetWebhooksParams, QuicknodeSdk};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    let webhooks = qn
        .webhooks
        .list_webhooks(&GetWebhooksParams {
            limit: 部分(10)，
            偏移量: 部分(0),
        })
        .await？;

    用于 webhook 在 webhooks.data {
        println!("{} {} {}", webhook.id, webhook.name, webhook.status);
    }

    let 已启用 = qn.webhooks.get_enabled_count().await?;
    println!("Enabled webhooks: {}", enabled.total);

    好的(())
}

Ruby

require "json"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

webhooks = qn.webhooks.list_webhooks({ limit: 10, offset: 0 })
webhooks["数据"].每个 do |webhook|
  puts "#{webhook["id"]} #{webhook["名称"]} #{webhook["状态"]}"
结束

已启用 = qn.webhooks.获取启用数量
puts "Enabled webhooks: #{enabled["total"]}"

有关模板行为和目标配置，请参阅Webhooks REST API 文档。

键值存储

使用键值存储客户端来实现简单的持久化状态。Set 用于在每个键下存储一个字符串值。List 用于在单个键下存储有序的字符串集合。

Node.js

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

const sets = await qn.kvstore.getSets({ limit: 10 });
console.log(`Sets: ${sets.data.length}`);

如果 (设置.数据.length > 0) {
  const set = await qn.kvstore.getSet(sets.data[0].键);
  控制台.log(set.value);
}

const lists = await qn.kvstore.getLists({ limit: 10 });
console.log(`Lists: ${lists.data.keys.length}`);

Python

import asyncio
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    集合 = await qn.kvstore.get_sets(limit=10)
    print(f"Sets: {len(sets.data)}")

    如果 len(设置.data) > 0:
        set_value = await qn.kvstore.get_set(集合.data[0].键)
        打印(set_value.value)

    列表 = await qn.kvstore.get_lists(limit=10)
    print(f"Lists: {len(lists.data.keys)}")

asyncio.run(main()))

Rust

use quicknode_sdk::{
    kvstore::{GetListsParams, GetSetsParams},
    QuicknodeSdk,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    let 集合 = qn
        .kvstore
        .get_sets(&GetSetsParams {
            limit: 部分(10)，
            光标: 无,
        })
        .await？;
    println!("Sets: {}", sets.data.len());

    如果 let 一些(first_set) = 集合.数据.第一() {
        let set = qn.kvstore.get_set(&first_set.key).await?;
        println!("{}", set.value);
    }

    let 列表 = qn
        .kvstore
        .get_lists(&GetListsParams {
            limit: 部分(10)，
            光标: 无,
        })
        .await？;
    println!("Lists: {}", lists.data.keys.len());

    好的(())
}

Ruby

require "json"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

sets = qn.kvstore.get_sets({ limit: 10 })
puts "Sets: #{sets["data"].length}"

除非 设置["data"]。空？
  set = qn.kvstore.get_set(key: 集合["data"][0]["key"])
  puts set["value"]
结束

lists = qn.kvstore.get_lists({ limit: 10 })
puts "Lists: #{lists["data"]["keys"].length}"

有关所有列表和集合操作，请参阅键值存储 REST API 文档。

SQL 资源管理器

使用 SQL Explorer 客户端对已建立索引的区块链数据运行 SQL 查询，并检索表结构。

Node.js

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

// Run a SQL query
const 结果 = await qn.sql.查询(
  "SELECT action_type, user FROMhyperliquid_system_actions LIMIT 3",
  "hyperliquid-core-mainnet"
);
console.log(`${result.rows} rows, ${result.credits} credits`);

// Get the schema for a cluster
const 模式 = await qn.sql.getSchema("hyperliquid-core-mainnet");
控制台.log(模式);

Python

import asyncio
来源quicknode_sdk import QuicknodeSdk

async def main():
    qn = QuicknodeSdk.from_env()

    # 执行 SQL 查询
    结果 = await qn.sql.查询(
        "SELECT action_type, user FROMhyperliquid_system_actions LIMIT 3",
        "hyperliquid-core-mainnet",
    )
    打印(f"{结果.行} 行数, {result.credits} 鸣谢")

    # 获取集群的架构
    模式 = await qn.sql.get_schema("hyperliquid-core-mainnet")
    打印(模式)

asyncio.run(main()))

Rust

使用 quicknode_sdk::QuicknodeSdk;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let qn = QuicknodeSdk::from_env()?;

    // Run a SQL query
    let 结果 = qn
        .sql
        .查询(
            "SELECT action_type, user FROMhyperliquid_system_actions LIMIT 3",
            "hyperliquid-core-mainnet",
        )
        .await？;
    println!("{} rows, {} credits", result.rows, result.credits);

    // Get the schema for a cluster
    let 模式 = qn.sql.get_schema("hyperliquid-core-mainnet").await？;
    println!("{:?}", schema);

    好的(())
}

Ruby

require "json"
require "quicknode_sdk"

qn = QuicknodeSdk::SDK.from_env

# 执行 SQL 查询
结果 = qn.sql.查询(
    "SELECT action_type, user FROMhyperliquid_system_actions LIMIT 3",
    "hyperliquid-core-mainnet",
)
puts "#{结果["行数"]} 行数, #{结果["credits"]} credits"

# 获取集群的架构
模式 = qn.sql.get_schema("hyperliquid-core-mainnet")
puts 模式

有关查询语法和可用表的信息，请参阅SQL Explorer 文档。

Agent 工作流示例

Agents 可以在单次运行中整合多个客户端。例如，发布自动化agent 可以在决定采取何种行动之前，检查端点、Streams 、Webhooks 、键值存储的状态，并查询区块链数据。

import { QuicknodeSdk } from "@quicknode/sdk";

const qn = QuicknodeSdk.fromEnv();

const endpoints = await qn.admin.getEndpoints({ tagLabels: ["sdk-managed"] });
const streams = await qn.streams.listStreams({ limit: 10 });
const webhooks = await qn.webhooks.listWebhooks({ limit: 10 });
const sets = await qn.kvstore.getSets({ limit: 10 });
const 交易 = await qn.sql.查询(
  "SELECT * FROMhyperliquid_trades LIMIT 5",
  "hyperliquid-core-mainnet"
);

控制台.log({
  端点: 端点.数据.长度,
  streams: streams.数据.长度,
  webhooks: webhooks.数据.长度,
  集合: 集合.数据.length,
  交易: 交易.行,
});