gRPC Client

Complete guide for integrating with NCN Network using gRPC.

---

Overview

gRPC provides:

• High performance binary protocol

• Streaming support

• Strong typing via Protocol Buffers

• Auto-generated client code

---

Generate Client Code

Python

# Install dependencies
pip install grpcio grpcio-tools

# Generate Python code
python -m grpc_tools.protoc \
  -I./proto \
  --python_out=./client \
  --grpc_python_out=./client \
  proto/common_types.proto

Go

# Install protoc-gen-go
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Generate Go code
protoc \
  --go_out=./client \
  --go-grpc_out=./client \
  proto/common_types.proto

JavaScript/TypeScript

# Install dependencies
npm install @grpc/grpc-js @grpc/proto-loader

# Or generate static code
npm install grpc-tools
grpc_tools_node_protoc \
  --js_out=import_style=commonjs:./client \
  --grpc_out=./client \
  proto/common_types.proto

Rust

Rust clients are built-in. Use the common_types crate:

[dependencies]
common_types = { path = "../common_types" }
tonic = "0.10"
tokio = { version = "1", features = ["full"] }

---

Connection Setup

Python

import grpc
from common_types_pb2 import InferenceRequest, TaskID
from common_types_pb2_grpc import GatewayClientServiceStub

# Create channel
channel = grpc.insecure_channel('localhost:50051')

# Or with TLS
credentials = grpc.ssl_channel_credentials()
channel = grpc.secure_channel('gateway.ncn-network.io:443', credentials)

# Create stub
stub = GatewayClientServiceStub(channel)

Go

import (
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    pb "your-module/proto"
)

// Insecure connection
conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
if err != nil {
    log.Fatal(err)
}
defer conn.Close()

// Or with TLS
creds, _ := credentials.NewClientTLSFromFile("ca.crt", "")
conn, err := grpc.Dial("gateway.ncn-network.io:443", grpc.WithTransportCredentials(creds))

// Create client
client := pb.NewGatewayClientServiceClient(conn)

JavaScript

const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');

// Load proto
const packageDef = protoLoader.loadSync('./proto/common_types.proto');
const proto = grpc.loadPackageDefinition(packageDef);

// Create client
const client = new proto.common_types.GatewayClientService(
  'localhost:50051',
  grpc.credentials.createInsecure()
);

Rust

use common_types::gateway_client_service_client::GatewayClientServiceClient;
use tonic::transport::Channel;

// Connect
let channel = Channel::from_static("http://localhost:50051")
    .connect()
    .await?;

// Create client
let mut client = GatewayClientServiceClient::new(channel);

---

Submit Inference Request

Python

import uuid

# Create request
request = InferenceRequest(
    request_id=str(uuid.uuid4()),
    model_uuid="bark_semantic",
    input_data='{"text": "Hello, world!"}',
    originator_client_id="client-123"
)

# Submit task
status = stub.SubmitInferenceTask(request)
print(f"Task ID: {status.request_id}")
print(f"Status: {status.status}")

Go

import (
    "context"
    "github.com/google/uuid"
)

ctx := context.Background()

request := &pb.InferenceRequest{
    RequestId:          uuid.New().String(),
    ModelUuid:          "bark_semantic",
    InputData:          `{"text": "Hello, world!"}`,
    OriginatorClientId: "client-123",
}

status, err := client.SubmitInferenceTask(ctx, request)
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Task ID: %s, Status: %s\n", status.RequestId, status.Status)

JavaScript

const { v4: uuidv4 } = require('uuid');

const request = {
  request_id: uuidv4(),
  model_uuid: 'bark_semantic',
  input_data: JSON.stringify({ text: 'Hello, world!' }),
  originator_client_id: 'client-123'
};

client.SubmitInferenceTask(request, (err, status) => {
  if (err) {
    console.error(err);
    return;
  }
  console.log(`Task ID: ${status.request_id}, Status: ${status.status}`);
});

Rust

use common_types::InferenceRequest;
use uuid::Uuid;

let request = InferenceRequest {
    request_id: Uuid::new_v4().to_string(),
    model_uuid: "bark_semantic".to_string(),
    input_data: r#"{"text": "Hello, world!"}"#.to_string(),
    originator_client_id: "client-123".to_string(),
    ..Default::default()
};

let status = client.submit_inference_task(request).await?.into_inner();
println!("Task ID: {}, Status: {}", status.request_id, status.status);

---

Subscribe to Results (Streaming)

Python

# Subscribe to updates
task_id = TaskID(id=status.request_id)

for response in stub.SubscribeToTaskUpdates(task_id):
    print(f"Status: {response.status}")
    
    if response.status == "completed":
        print(f"Output: {response.output_data}")
        break
    elif response.status == "failed":
        print(f"Error: {response.error_message}")
        break

Go

stream, err := client.SubscribeToTaskUpdates(ctx, &pb.TaskID{Id: status.RequestId})
if err != nil {
    log.Fatal(err)
}

for {
    response, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("Status: %s\n", response.Status)
    
    if response.Status == "completed" {
        fmt.Printf("Output: %s\n", response.OutputData)
        break
    } else if response.Status == "failed" {
        fmt.Printf("Error: %s\n", response.ErrorMessage)
        break
    }
}

JavaScript

const call = client.SubscribeToTaskUpdates({ id: status.request_id });

call.on('data', (response) => {
  console.log(`Status: ${response.status}`);
  
  if (response.status === 'completed') {
    console.log(`Output: ${response.output_data}`);
    call.cancel();
  } else if (response.status === 'failed') {
    console.log(`Error: ${response.error_message}`);
    call.cancel();
  }
});

call.on('end', () => {
  console.log('Stream ended');
});

call.on('error', (err) => {
  console.error(err);
});

Rust

use common_types::TaskId;
use tokio_stream::StreamExt;

let task_id = TaskId { id: status.request_id.clone() };
let mut stream = client.subscribe_to_task_updates(task_id).await?.into_inner();

while let Some(response) = stream.next().await {
    let response = response?;
    println!("Status: {}", response.status);
    
    if response.status == "completed" {
        println!("Output: {}", response.output_data);
        break;
    } else if response.status == "failed" {
        println!("Error: {}", response.error_message);
        break;
    }
}

---

Error Handling

gRPC Status Codes

Code	Name	Description
0	OK	Success
1	CANCELLED	Operation cancelled
3	INVALID_ARGUMENT	Bad request
5	NOT_FOUND	Resource not found
8	RESOURCE_EXHAUSTED	Rate limited
13	INTERNAL	Server error
14	UNAVAILABLE	Service unavailable

Python Error Handling

import grpc

try:
    status = stub.SubmitInferenceTask(request)
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.UNAVAILABLE:
        print("Service unavailable, retrying...")
    elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
        print(f"Invalid request: {e.details()}")
    else:
        print(f"Error: {e.code()} - {e.details()}")

Go Error Handling

import "google.golang.org/grpc/status"

status, err := client.SubmitInferenceTask(ctx, request)
if err != nil {
    st, ok := status.FromError(err)
    if ok {
        switch st.Code() {
        case codes.Unavailable:
            log.Println("Service unavailable, retrying...")
        case codes.InvalidArgument:
            log.Printf("Invalid request: %s", st.Message())
        default:
            log.Printf("Error: %s - %s", st.Code(), st.Message())
        }
    }
}

---

Retry Logic

Python with Retry

import time
from functools import wraps

def retry(max_retries=3, delay=1.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except grpc.RpcError as e:
                    if e.code() in [grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.RESOURCE_EXHAUSTED]:
                        if attempt < max_retries - 1:
                            time.sleep(delay * (2 ** attempt))
                            continue
                    raise
        return wrapper
    return decorator

@retry(max_retries=3, delay=1.0)
def submit_task(stub, request):
    return stub.SubmitInferenceTask(request)

---

Connection Pooling

Python

# Use channel pool for high throughput
class ChannelPool:
    def __init__(self, target, pool_size=10):
        self.channels = [
            grpc.insecure_channel(target)
            for _ in range(pool_size)
        ]
        self.index = 0
    
    def get_stub(self):
        channel = self.channels[self.index]
        self.index = (self.index + 1) % len(self.channels)
        return GatewayClientServiceStub(channel)

pool = ChannelPool('localhost:50051', pool_size=10)
stub = pool.get_stub()

---

Full Example

Python Complete Client

#!/usr/bin/env python3
import grpc
import uuid
import json
from common_types_pb2 import InferenceRequest, TaskID
from common_types_pb2_grpc import GatewayClientServiceStub

class NCNClient:
    def __init__(self, gateway_url='localhost:50051'):
        self.channel = grpc.insecure_channel(gateway_url)
        self.stub = GatewayClientServiceStub(self.channel)
    
    def inference(self, model_uuid, input_data, timeout=60):
        """Submit inference and wait for result."""
        # Create request
        request = InferenceRequest(
            request_id=str(uuid.uuid4()),
            model_uuid=model_uuid,
            input_data=json.dumps(input_data) if isinstance(input_data, dict) else input_data,
            originator_client_id="python-client"
        )
        
        # Submit
        status = self.stub.SubmitInferenceTask(request)
        print(f"Submitted: {status.request_id}")
        
        # Wait for result
        for response in self.stub.SubscribeToTaskUpdates(
            TaskID(id=status.request_id),
            timeout=timeout
        ):
            if response.status == "completed":
                return json.loads(response.output_data)
            elif response.status == "failed":
                raise Exception(response.error_message)
        
        raise TimeoutError("Request timed out")
    
    def close(self):
        self.channel.close()

# Usage
if __name__ == "__main__":
    client = NCNClient()
    try:
        result = client.inference(
            model_uuid="bark_semantic",
            input_data={"text": "Hello, world!"}
        )
        print(f"Result: {result}")
    finally:
        client.close()

---

Next Steps

• HTTP Client - REST API integration

• Payment Integration - Handle payments

• Examples - More code examples