Skip to content

Indexer API Reference

The Indexer module provides an interface for querying the Midnight GraphQL indexer to fetch blocks, transactions, and UTXOs.

Indexer Client

noxipher.indexer.client.IndexerClient

GraphQL client for Midnight Indexer v4.

Supports: - HTTP queries (block, transaction, UTXO, DUST) - WebSocket subscriptions (blocks, shielded transactions, ZSwap events)

Source code in src/noxipher/indexer/client.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class IndexerClient:
    """
    GraphQL client for Midnight Indexer v4.

    Supports:
    - HTTP queries (block, transaction, UTXO, DUST)
    - WebSocket subscriptions (blocks, shielded transactions, ZSwap events)
    """

    def __init__(self, config: NetworkConfig) -> None:
        self._http_url = config.indexer_http_url
        self._ws_url = config.indexer_ws_url
        self._http_client: Client | None = None
        self._ws_client: Client | None = None

    async def __aenter__(self) -> IndexerClient:
        transport = AIOHTTPTransport(url=self._http_url)
        self._http_client = Client(transport=transport, fetch_schema_from_transport=False)
        await self._http_client.__aenter__()  # type: ignore[no-untyped-call]
        return self

    async def __aexit__(self, *args: object) -> None:
        if self._http_client:
            await self._http_client.__aexit__(*args)  # type: ignore[no-untyped-call]

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
    async def get_block(self, height: int | None = None, hash_hex: str | None = None) -> Block:
        """Get block by height or hash. Default: latest block."""
        assert self._http_client is not None
        try:
            result = await self._http_client.execute_async(
                gql(GET_BLOCK),
                variable_values={"height": height, "hash": hash_hex},
            )
            return Block.model_validate(result["block"])

        except Exception as e:
            raise IndexerError(f"get_block failed: {e}") from e

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
    async def get_transactions(
        self,
        hash: str | None = None,
        address: str | None = None,
        limit: int = 10,
    ) -> list[Transaction]:
        """Get transactions by hash or address."""
        assert self._http_client is not None
        try:
            result = await self._http_client.execute_async(
                gql(GET_TRANSACTIONS),
                variable_values={"hash": hash, "address": address, "limit": limit},
            )
            data = result
            return [Transaction.model_validate(tx) for tx in data["transactions"]["nodes"]]

        except Exception as e:
            raise IndexerError(f"get_transactions failed: {e}") from e

    async def connect_wallet_session(self, viewing_key: str) -> str:
        """
        Open shielded wallet session with Indexer.

        Viewing key format: hex-encoded (coinPublicKey + encryptionPublicKey) — 64 bytes hex
        Returns: session_id string
        """
        mutation = gql("""
            mutation ConnectWallet($viewingKey: ViewingKey!) {
                connect(viewingKey: $viewingKey)
            }
        """)
        assert self._http_client is not None
        try:
            result = await self._http_client.execute_async(
                mutation,
                variable_values={"viewingKey": viewing_key},
            )
            return cast(str, result["connect"])

        except Exception as e:
            raise IndexerError(f"connect_wallet_session failed: {e}") from e

    async def disconnect_wallet_session(self, session_id: str) -> None:
        """Close shielded wallet session."""
        mutation = gql("""
            mutation DisconnectWallet($sessionId: String!) {
                disconnect(sessionId: $sessionId)
            }
        """)
        assert self._http_client is not None
        try:
            await self._http_client.execute_async(
                mutation,
                variable_values={"sessionId": session_id},
            )

        except Exception as e:
            raise IndexerError(f"disconnect_wallet_session failed: {e}") from e

    async def subscribe_blocks(self) -> AsyncIterator[Block]:
        """Subscribe to new blocks via WebSocket."""
        transport = WebsocketsTransport(url=self._ws_url)
        async with Client(transport=transport) as session:
            async for result in session.subscribe(gql(SUBSCRIBE_BLOCKS)):
                yield Block.model_validate(result["blockAdded"])

    async def subscribe_shielded_transactions(
        self, session_id: str
    ) -> AsyncIterator[dict[str, Any]]:
        """
        Stream shielded transactions for wallet session.
        Returns ShieldedTransactionFound + ShieldedTransactionProgress events.
        """
        transport = WebsocketsTransport(url=self._ws_url)
        async with Client(transport=transport) as session:
            async for result in session.subscribe(
                gql(SUBSCRIBE_SHIELDED_TXS),
                variable_values={"sessionId": session_id},
            ):
                yield result["shieldedTransaction"]

    async def get_dust_status(self, cardano_stake_keys: list[str]) -> list[DustGenerationStatus]:
        """Query DUST generation status for Cardano stake keys."""
        assert self._http_client is not None
        result = await self._http_client.execute_async(
            gql(GET_DUST_STATUS),
            variable_values={"stakeKeys": cardano_stake_keys},
        )
        data = result
        return [DustGenerationStatus.model_validate(d) for d in data["dustStatus"]]

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
    async def get_utxos(self, address: str) -> list[dict[str, Any]]:
        """Get unshielded UTXOs for an address."""
        assert self._http_client is not None
        try:
            result = await self._http_client.execute_async(
                gql(GET_UTXOS),
                variable_values={"address": address},
            )
            data = result
            return cast(list[dict[str, Any]], data["unshieldedUtxos"]["nodes"])

        except Exception as e:
            raise IndexerError(f"get_utxos failed: {e}") from e

connect_wallet_session(viewing_key) async

Open shielded wallet session with Indexer.

Viewing key format: hex-encoded (coinPublicKey + encryptionPublicKey) — 64 bytes hex Returns: session_id string

Source code in src/noxipher/indexer/client.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def connect_wallet_session(self, viewing_key: str) -> str:
    """
    Open shielded wallet session with Indexer.

    Viewing key format: hex-encoded (coinPublicKey + encryptionPublicKey) — 64 bytes hex
    Returns: session_id string
    """
    mutation = gql("""
        mutation ConnectWallet($viewingKey: ViewingKey!) {
            connect(viewingKey: $viewingKey)
        }
    """)
    assert self._http_client is not None
    try:
        result = await self._http_client.execute_async(
            mutation,
            variable_values={"viewingKey": viewing_key},
        )
        return cast(str, result["connect"])

    except Exception as e:
        raise IndexerError(f"connect_wallet_session failed: {e}") from e

disconnect_wallet_session(session_id) async

Close shielded wallet session.

Source code in src/noxipher/indexer/client.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def disconnect_wallet_session(self, session_id: str) -> None:
    """Close shielded wallet session."""
    mutation = gql("""
        mutation DisconnectWallet($sessionId: String!) {
            disconnect(sessionId: $sessionId)
        }
    """)
    assert self._http_client is not None
    try:
        await self._http_client.execute_async(
            mutation,
            variable_values={"sessionId": session_id},
        )

    except Exception as e:
        raise IndexerError(f"disconnect_wallet_session failed: {e}") from e

get_block(height=None, hash_hex=None) async

Get block by height or hash. Default: latest block.

Source code in src/noxipher/indexer/client.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def get_block(self, height: int | None = None, hash_hex: str | None = None) -> Block:
    """Get block by height or hash. Default: latest block."""
    assert self._http_client is not None
    try:
        result = await self._http_client.execute_async(
            gql(GET_BLOCK),
            variable_values={"height": height, "hash": hash_hex},
        )
        return Block.model_validate(result["block"])

    except Exception as e:
        raise IndexerError(f"get_block failed: {e}") from e

get_dust_status(cardano_stake_keys) async

Query DUST generation status for Cardano stake keys.

Source code in src/noxipher/indexer/client.py 
147
148
149
150
151
152
153
154
155
async def get_dust_status(self, cardano_stake_keys: list[str]) -> list[DustGenerationStatus]:
    """Query DUST generation status for Cardano stake keys."""
    assert self._http_client is not None
    result = await self._http_client.execute_async(
        gql(GET_DUST_STATUS),
        variable_values={"stakeKeys": cardano_stake_keys},
    )
    data = result
    return [DustGenerationStatus.model_validate(d) for d in data["dustStatus"]]

get_transactions(hash=None, address=None, limit=10) async

Get transactions by hash or address.

Source code in src/noxipher/indexer/client.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def get_transactions(
    self,
    hash: str | None = None,
    address: str | None = None,
    limit: int = 10,
) -> list[Transaction]:
    """Get transactions by hash or address."""
    assert self._http_client is not None
    try:
        result = await self._http_client.execute_async(
            gql(GET_TRANSACTIONS),
            variable_values={"hash": hash, "address": address, "limit": limit},
        )
        data = result
        return [Transaction.model_validate(tx) for tx in data["transactions"]["nodes"]]

    except Exception as e:
        raise IndexerError(f"get_transactions failed: {e}") from e

get_utxos(address) async

Get unshielded UTXOs for an address.

Source code in src/noxipher/indexer/client.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
async def get_utxos(self, address: str) -> list[dict[str, Any]]:
    """Get unshielded UTXOs for an address."""
    assert self._http_client is not None
    try:
        result = await self._http_client.execute_async(
            gql(GET_UTXOS),
            variable_values={"address": address},
        )
        data = result
        return cast(list[dict[str, Any]], data["unshieldedUtxos"]["nodes"])

    except Exception as e:
        raise IndexerError(f"get_utxos failed: {e}") from e

subscribe_blocks() async

Subscribe to new blocks via WebSocket.

Source code in src/noxipher/indexer/client.py 
125
126
127
128
129
130
async def subscribe_blocks(self) -> AsyncIterator[Block]:
    """Subscribe to new blocks via WebSocket."""
    transport = WebsocketsTransport(url=self._ws_url)
    async with Client(transport=transport) as session:
        async for result in session.subscribe(gql(SUBSCRIBE_BLOCKS)):
            yield Block.model_validate(result["blockAdded"])

subscribe_shielded_transactions(session_id) async

Stream shielded transactions for wallet session. Returns ShieldedTransactionFound + ShieldedTransactionProgress events.

Source code in src/noxipher/indexer/client.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def subscribe_shielded_transactions(
    self, session_id: str
) -> AsyncIterator[dict[str, Any]]:
    """
    Stream shielded transactions for wallet session.
    Returns ShieldedTransactionFound + ShieldedTransactionProgress events.
    """
    transport = WebsocketsTransport(url=self._ws_url)
    async with Client(transport=transport) as session:
        async for result in session.subscribe(
            gql(SUBSCRIBE_SHIELDED_TXS),
            variable_values={"sessionId": session_id},
        ):
            yield result["shieldedTransaction"]

Indexer Models

noxipher.indexer.models

Indexer data models — Pydantic models for Midnight Indexer v4 responses.

Block

Bases: BaseModel

Block data from Indexer.

Source code in src/noxipher/indexer/models.py 
12
13
14
15
16
17
18
class Block(BaseModel):
    """Block data from Indexer."""

    height: int
    hash: str
    parent_hash: str | None = None
    timestamp: str | None = None

DustGenerationStatus

Bases: BaseModel

DUST generation status for a Cardano stake key.

Source code in src/noxipher/indexer/models.py 
54
55
56
57
58
59
60
class DustGenerationStatus(BaseModel):
    """DUST generation status for a Cardano stake key."""

    cardano_stake_key: str
    is_registered: bool
    available_dust: str  # Bigint as string (Specks)
    registered_utxos: list[dict[str, Any]] = []

TokenBalance

Bases: BaseModel

Token balance from Indexer.

Source code in src/noxipher/indexer/models.py 
34
35
36
37
38
class TokenBalance(BaseModel):
    """Token balance from Indexer."""

    token_type: str  # hex-encoded RawTokenType
    value: str  # Bigint as string (Specks)

Transaction

Bases: BaseModel

Transaction data from Indexer.

Source code in src/noxipher/indexer/models.py 
41
42
43
44
45
46
47
48
49
50
51
class Transaction(BaseModel):
    """Transaction data from Indexer."""

    hash: str
    block: Block | None = None
    transaction_result: TransactionResult | None = None
    fees: dict[str, Any] | None = None
    raw: str | None = None  # HexEncoded raw bytes
    unshielded_created_outputs: list[dict[str, Any]] = []
    unshielded_spent_outputs: list[dict[str, Any]] = []
    events: list[dict[str, Any]] = []

TransactionResult

Bases: BaseModel

Transaction execution result.

Source code in src/noxipher/indexer/models.py 
21
22
23
24
25
26
27
28
29
30
31
class TransactionResult(BaseModel):
    """Transaction execution result."""

    status: str  # "success", "PARTIAL_SUCCESS", "failure"
    segments: list[dict[str, Any]] = []

    @field_validator("status", mode="before")
    @classmethod
    def normalize_status(cls, v: str) -> str:
        """Normalize status string — Indexer returns mixed case."""
        return v.upper() if isinstance(v, str) else v

normalize_status(v) classmethod

Normalize status string — Indexer returns mixed case.

Source code in src/noxipher/indexer/models.py 
27
28
29
30
31
@field_validator("status", mode="before")
@classmethod
def normalize_status(cls, v: str) -> str:
    """Normalize status string — Indexer returns mixed case."""
    return v.upper() if isinstance(v, str) else v