gRPC API Reference¶
PrimeStamp provides a high-performance gRPC API for document stamping with support for unary RPCs, server streaming, client streaming, and bidirectional streaming.
Setup¶
Starting the Server¶
from primestamp.api.grpc_api import create_grpc_server
server = create_grpc_server(host="0.0.0.0", port=50051)
await server.start()
await server.wait_for_termination()
Creating a Client¶
from primestamp.api.grpc_api import create_grpc_client
client = create_grpc_client(host="localhost", port=50051)
await client.connect()
Service Definition¶
syntax = "proto3";
package primestamp.v1;
import "google/protobuf/timestamp.proto";
service PrimeStampService {
// Unary RPCs
rpc CreateStamp(CreateStampRequest) returns (CreateStampResponse);
rpc GetStamp(GetStampRequest) returns (GetStampResponse);
rpc Verify(VerifyRequest) returns (VerifyResponse);
rpc BatchStamp(BatchStampRequest) returns (BatchStampResponse);
// Server streaming
rpc WatchStamps(WatchStampsRequest) returns (stream Stamp);
// Client streaming
rpc StreamStamps(stream StreamStampRequest) returns (stream StreamStampResponse);
// Health check
rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
}
Enums¶
enum HashAlgorithm {
HASH_ALGORITHM_UNSPECIFIED = 0;
HASH_ALGORITHM_SHA3_256 = 1;
HASH_ALGORITHM_SHA256 = 2;
}
enum StampStatus {
STAMP_STATUS_UNSPECIFIED = 0;
STAMP_STATUS_PENDING = 1;
STAMP_STATUS_CONFIRMED = 2;
STAMP_STATUS_ANCHORED = 3;
}
Messages¶
Stamp¶
message Stamp {
string id = 1;
string content_hash = 2;
HashAlgorithm algorithm = 3;
google.protobuf.Timestamp timestamp = 4;
StampStatus status = 5;
string description = 6;
map<string, string> metadata = 7;
Proof proof = 8;
}
Proof¶
message Proof {
string prime_signature = 1;
repeated string witnesses = 2;
string chain_anchor = 3;
string merkle_root = 4;
repeated string merkle_proof = 5;
}
Request/Response Messages¶
message CreateStampRequest {
string content_hash = 1;
HashAlgorithm algorithm = 2;
string description = 3;
map<string, string> metadata = 4;
}
message CreateStampResponse {
Stamp stamp = 1;
}
message GetStampRequest {
string stamp_id = 1;
}
message GetStampResponse {
Stamp stamp = 1;
}
message VerifyRequest {
string stamp_id = 1;
string content_hash = 2;
}
message VerifyResponse {
bool verified = 1;
string stamp_id = 2;
string content_hash = 3;
google.protobuf.Timestamp timestamp = 4;
string message = 5;
}
message BatchStampRequest {
repeated CreateStampRequest items = 1;
}
message BatchStampResponse {
repeated Stamp stamps = 1;
int32 success_count = 2;
int32 failed_count = 3;
}
message StreamStampRequest {
string content_hash = 1;
map<string, string> metadata = 2;
}
message StreamStampResponse {
Stamp stamp = 1;
int32 sequence = 2;
}
message WatchStampsRequest {
string filter = 1;
}
message HealthCheckRequest {
string service = 1;
}
message HealthCheckResponse {
enum ServingStatus {
UNKNOWN = 0;
SERVING = 1;
NOT_SERVING = 2;
}
ServingStatus status = 1;
}
RPC Methods¶
CreateStamp (Unary)¶
Create a single stamp.
from primestamp.api.grpc_api import CreateStampRequest, HashAlgorithm
request = CreateStampRequest(
content_hash="sha3-256:e3b0c44298fc1c14...",
algorithm=HashAlgorithm.HASH_ALGORITHM_SHA3_256,
description="Quarterly report",
metadata={"department": "finance"},
)
response = await servicer.CreateStamp(request)
print(response.stamp.id)
GetStamp (Unary)¶
Retrieve a stamp by ID.
from primestamp.api.grpc_api import GetStampRequest
request = GetStampRequest(stamp_id="ps-abc123...")
response = await servicer.GetStamp(request)
print(response.stamp.content_hash)
Verify (Unary)¶
Verify content against a stored stamp.
from primestamp.api.grpc_api import VerifyRequest
request = VerifyRequest(
stamp_id="ps-abc123...",
content_hash="sha3-256:e3b0c44298fc1c14...",
)
response = await servicer.Verify(request)
print(f"Verified: {response.verified}, Message: {response.message}")
BatchStamp (Unary)¶
Create multiple stamps in a single call.
from primestamp.api.grpc_api import BatchStampRequest, CreateStampRequest
request = BatchStampRequest(items=[
CreateStampRequest(content_hash="sha3-256:aaa..."),
CreateStampRequest(content_hash="sha3-256:bbb..."),
CreateStampRequest(content_hash="sha3-256:ccc..."),
])
response = await servicer.BatchStamp(request)
print(f"Created {response.success_count} stamps")
WatchStamps (Server Streaming)¶
Subscribe to a stream of new stamps from the server.
request = {"filter": ""}
async for stamp in servicer.WatchStamps(request):
print(f"New stamp: {stamp.id} - {stamp.content_hash}")
StreamStamps (Bidirectional Streaming)¶
Stream stamp creation requests to the server and receive responses for each.
from primestamp.api.grpc_api import StreamStampRequest
async def request_generator():
for hash_val in hashes:
yield StreamStampRequest(
content_hash=hash_val,
metadata={"source": "pipeline"},
)
async for response in servicer.StreamStamps(request_generator()):
print(f"Stamp #{response.sequence}: {response.stamp.id}")
HealthCheck (Unary)¶
from primestamp.api.grpc_api import HealthCheckRequest
response = await servicer.HealthCheck(HealthCheckRequest(service="primestamp"))
print(response.status) # "SERVING"
Server Lifecycle¶
import asyncio
from primestamp.api.grpc_api import create_grpc_server
async def main():
server = create_grpc_server(host="0.0.0.0", port=50051)
await server.start()
print("gRPC server listening on port 50051")
try:
await server.wait_for_termination()
except KeyboardInterrupt:
await server.stop(grace=5.0)
asyncio.run(main())
Generating Python Stubs¶
If you need compiled protobuf stubs, run:
python -m grpc_tools.protoc \
-I. \
--python_out=. \
--grpc_python_out=. \
primestamp/api/primestamp.proto
The proto definition is also available programmatically: