Developer Platform: gRPC & GraphQL

15.1 Why a Developer API Layer

The REST Query API (POST /queries/{name}/execute) is purpose-built for interactive dashboards and single-shot queries. When you need to build integrations — data pipelines, custom dashboards, downstream CMDB sync — the REST API's per-request overhead and JSON-over-HTTP/1.1 model becomes a bottleneck.

The Developer Platform adds two purpose-built interfaces:

• gRPC QueryService — server-side streaming over HTTP/2; ideal for streaming large result sets (10,000-row topology exports), real-time data pipelines, and language-idiomatic clients (Python, Go, Java, Rust).
• GraphQL API — single schema, interactive IDE, field selection; ideal for front-end teams building custom dashboards or tooling that needs query introspection.

Performance Comparison

15.2 gRPC QueryService

Pro+ License gate: api_access

Installation

pip install 'meshoptixiq-network-discovery[grpc]'
# Installs grpcio>=1.60 and protobuf>=4.0

Startup

# Start the gRPC server (default port 8010)
meshq-grpc

# Override the port
MESHQ_GRPC_PORT=9010 meshq-grpc

Service Definition

syntax = "proto3";

service QueryService {
  // List all registered queries (no auth required)
  rpc ListQueries(ListQueriesRequest) returns (ListQueriesResponse);

  // Get metadata for a single query
  rpc GetQuery(GetQueryRequest) returns (QueryInfo);

  // Execute a query and stream results back as RowBatch messages
  rpc ExecuteQuery(ExecuteQueryRequest) returns (stream RowBatch);
}

Key Message Types

Authentication

Pass credentials as gRPC metadata on every call:

import grpc

# Option 1: API key header
metadata = [('x-api-key', 'your-api-key')]

# Option 2: Bearer token (PAT)
metadata = [('authorization', 'Bearer mq_live_your-token-here')]

Python Client Example

import grpc
from network_discovery.grpc import query_pb2, query_pb2_grpc

def stream_all_devices(host: str = "localhost:8010", api_key: str = "demo"):
    """Stream all devices from the graph via gRPC."""
    channel = grpc.insecure_channel(host)
    stub = query_pb2_grpc.QueryServiceStub(channel)

    request = query_pb2.ExecuteQueryRequest(
        name="all_devices",
        parameters={},
        limit=10000,
        offset=0,
    )
    metadata = [("x-api-key", api_key)]

    rows = []
    for batch in stub.ExecuteQuery(request, metadata=metadata):
        rows.extend(batch.rows)
        print(f"Received batch: {len(batch.rows)} rows "
              f"(total so far: {len(rows)} / {batch.total})")

    return rows

devices = stream_all_devices()
print(f"Total devices: {len(devices)}")

Pagination

The server streams 200 rows per RowBatch automatically. The total field in the first batch tells you the complete result set size. For results beyond the 10,000-row hard cap, use offset to paginate across multiple calls:

all_rows = []
page_size = 5000
offset = 0

while True:
    request = query_pb2.ExecuteQueryRequest(
        name="all_devices", parameters={}, limit=page_size, offset=offset
    )
    batch_rows = []
    total = 0
    for batch in stub.ExecuteQuery(request, metadata=metadata):
        batch_rows.extend(batch.rows)
        total = batch.total

    all_rows.extend(batch_rows)
    offset += page_size
    if offset >= total:
        break

Sequence Diagram

15.3 GraphQL API

Pro+ License gate: api_access

Installation

pip install 'meshoptixiq-network-discovery[graphql]'
# Installs strawberry-graphql>=0.220

Endpoint

Schema

type Query {
  # List all registered queries
  queries: [QueryInfo!]!

  # Get a single query by name
  query(name: String!): QueryInfo

  # Execute a named query and return results
  executeQuery(
    name: String!
    parameters: JSON
    limit: Int = 1000
    offset: Int = 0
  ): QueryResult!
}

type QueryResult {
  total: Int!
  rows: [JSON!]!
}

type QueryInfo {
  name: String!
  description: String!
  category: String!
  parameters: [QueryParameter!]!
  minTier: String!
}

type QueryParameter {
  name: String!
  type: String!
  required: Boolean!
}

Example Query

# Execute summary_stats and return the first 10 rows
{
  executeQuery(name: "summary_stats", parameters: {}) {
    total
    rows
  }
}

# Get all topology queries with their parameter schemas
{
  queries {
    name
    category
    parameters {
      name
      type
    }
  }
}

Calling from curl

curl -X POST http://localhost:8000/graphql \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{"query": "{ executeQuery(name: \"summary_stats\", parameters: {}) { total rows } }"}'

Authentication

Pass the same credentials as the REST API — either X-API-Key or Authorization: Bearer mq_live_<token> as an HTTP header. Auth failures return structured GraphQL errors (not HTTP 4xx):

{
  "data": null,
  "errors": [
    {
      "message": "Not authenticated",
      "locations": [{"line": 1, "column": 1}],
      "path": ["executeQuery"]
    }
  ]
}

15.4 Authentication & License Gating

Authentication Methods

License Gating

Demo Mode

Set MESHOPTIXIQ_DEMO_MODE=true to bypass all license checks. In demo mode, all interfaces are accessible without a license key. The API key defaults to demo and the in-memory graph backend is automatically selected.

MESHOPTIXIQ_DEMO_MODE=true GRAPH_BACKEND=inmemory API_KEY=demo meshq-grpc