MEV SDK — Python

The yoorquezt-sdk-mev package provides async Python clients for the Q AI API and the MEV Gateway.

Installation

pip install yoorquezt-sdk-mev
# or
poetry add yoorquezt-sdk-mev

QMEVClient

The QMEVClient connects to the Q AI API for natural language chat and tool access. All methods are async.

from yoorquezt_sdk_mev import QMEVClient

client = QMEVClient(
    base_url="http://localhost:9100",
    api_key="your-api-key",
    role="operator",
)

chat

Send a message and receive a complete response.

response = await client.chat("What's the profit from the last 24 hours?")

print(response.content)
# "Total profit (24h): 2.847 ETH across 142 landed bundles."

print(response.tools_used)
# ["analyst_profit"]

chat_stream_iter

Stream a response token-by-token using an async iterator.

async for event in client.chat_stream_iter("Analyze relay performance"):
    if event.type == "content_delta":
        print(event.delta, end="", flush=True)
    elif event.type == "tool_use":
        print(f"\n[Using tool: {event.tool}]")
    elif event.type == "tool_result":
        print(f"[Tool result: {event.tool}]")
    elif event.type == "confirmation_required":
        print(f"\nAction requires confirmation: {event.message}")
    elif event.type == "message_end":
        print(f"\n[Tokens: {event.usage.input_tokens} in, {event.usage.output_tokens} out]")

list_tools

List all tools available to your role.

tools = await client.list_tools()

for tool in tools:
    print(f"{tool.name} ({tool.category}) - {tool.description}")
    if tool.mutating:
        print("  [MUTATING - requires confirmation]")

health

Check Q AI API health.

health = await client.health()
print(health.status)  # "healthy"

MEVGatewayClient

The MEVGatewayClient connects to the MEV Gateway for direct engine interaction.

from yoorquezt_sdk_mev import MEVGatewayClient

gw = MEVGatewayClient(
    url="ws://localhost:9099",
    api_key="your-api-key",
)

submit_bundle

result = await gw.submit_bundle(
    txs=["0xsigned_tx_1", "0xsigned_tx_2"],
    block_number=19482300,
    chain="ethereum",
)

print(result.bundle_hash)  # "0x9f8e7d..."
print(result.status)       # "pending"

get_bundle_status

status = await gw.get_bundle_status("0x9f8e7d...")

print(status.status)   # "landed"
print(status.block)    # 19482301
print(status.profit)   # "0.0312 ETH"

get_mempool_snapshot

mempool = await gw.get_mempool_snapshot()

print(f"Pending transactions: {mempool.pending}")
for tx in mempool.txs[:5]:
    print(f"  {tx.hash} - {tx.value} ETH")

get_relay_stats

stats = await gw.get_relay_stats()

for relay in stats.relays:
    print(f"{relay.name}: {relay.status} ({relay.latency_ms}ms)")

get_profit_history

history = await gw.get_profit_history(
    range="7d",
    chain="ethereum",
    group_by="day",
)

for entry in history:
    print(f"{entry.date}: {entry.profit_eth} ETH ({entry.bundles} bundles)")

Pydantic Models

All response types are Pydantic models with full type hints and validation.

Utilities

The SDK includes formatting and conversion utilities:

from yoorquezt_sdk_mev.utils import format_wei, format_gwei, parse_hex

# Wei to ETH
print(format_wei(2847000000000000000))  # "2.847 ETH"

# Wei to Gwei
print(format_gwei(30000000000))  # "30.0 Gwei"

# Parse hex value
print(parse_hex("0x1BC16D674EC80000"))  # 2000000000000000000

Error Handling

The SDK raises QMEVError for all error conditions:

from yoorquezt_sdk_mev import QMEVError

try:
    await gw.get_bundle_status("0xinvalid")
except QMEVError as e:
    print(f"Error {e.code}: {e.message}")
    # Error 1002: Bundle not found

    if e.code == 1001:
        print("Engine unavailable, retry later")
    elif e.code == 1002:
        print("Resource not found")
    elif e.code == 1003:
        print(f"Rate limited, retry after {e.retry_after}s")

Full Example

import asyncio
from yoorquezt_sdk_mev import QMEVClient, MEVGatewayClient

async def main():
    # Q AI for natural language
    ai = QMEVClient(
        base_url="http://localhost:9100",
        api_key="your-api-key",
        role="operator",
    )

    # Gateway for direct engine access
    gw = MEVGatewayClient(
        url="ws://localhost:9099",
        api_key="your-api-key",
    )

    # Check health
    health = await ai.health()
    print(f"Q AI: {health.status}")

    # Ask Q AI about profit
    response = await ai.chat("Show me today's profit by strategy")
    print(response.content)

    # Stream a complex query
    print("\n--- Forensic Report ---")
    async for event in ai.chat_stream_iter(
        "Generate a forensic report for the last 100 blocks"
    ):
        if event.type == "content_delta":
            print(event.delta, end="", flush=True)
    print()

    # Direct bundle submission via gateway
    result = await gw.submit_bundle(
        txs=["0xsigned_tx_1", "0xsigned_tx_2"],
        block_number=19482300,
        chain="ethereum",
    )
    print(f"\nBundle submitted: {result.bundle_hash}")

    # Check status
    status = await gw.get_bundle_status(result.bundle_hash)
    print(f"Status: {status.status}")

    # Profit history
    history = await gw.get_profit_history(range="7d", chain="ethereum")
    for entry in history:
        print(f"  {entry.date}: {entry.profit_eth} ETH")

asyncio.run(main())