“

Series: S8 Technology Deep Dives · Article 15 | Level: Advanced | Reading Time: 20 min

FastAPI + gRPC Dual-Protocol Service: Intelligence Layer Python Microservice Architecture

#TL;DR

coomia-dip's Intelligence Layer (Reasoning & Decision Layer + Agent Runtime Layer) is built with Python 3.x, exposing external REST APIs via FastAPI and communicating with internal services via gRPC
This article deeply analyzes dual-protocol architecture design motivations, FastAPI's async/await model, gRPC Python Server threading model, Protobuf serialization optimization, and unified error handling across both protocols
Covers automatic Pydantic v2 to Protobuf Message conversion, health checks, middleware chains, and production deployment patterns

#1. Dual-Protocol Architecture Motivation

#1.1 Why Two Protocols?

Code

External Clients (Browser/SDK)           Internal Services (Control Layer + Data Layer)
      │                                     │
      │  REST/JSON (easy to debug, universal) │  gRPC/Protobuf (high perf, type-safe)
      │                                     │
      ▼                                     ▼
┌─────────────────────────────────────────────┐
│           Intelligence Layer (Python)        │
│                                              │
│  ┌──────────────┐    ┌──────────────────┐   │
│  │  FastAPI      │    │  gRPC Server     │   │
│  │  Port 8000    │    │  Port 50051      │   │
│  │  REST/JSON    │    │  Protobuf        │   │
│  └──────┬───────┘    └──────┬───────────┘   │
│         │                    │               │
│         └────────┬───────────┘               │
│                  │                           │
│         ┌────────┴────────┐                  │
│         │  Service Layer  │                  │
│         │ (shared logic)  │                  │
│         └─────────────────┘                  │
└──────────────────────────────────────────────┘

#1.2 Protocol Selection Matrix

Caller	Protocol	Rationale
Frontend/Browser	REST	Native browser support
Python SDK	REST	Simple and easy to use
Control Layer (Java)	gRPC	Type safety, high performance
Data Layer (Java)	gRPC	Streaming, low latency
Agent Runtime (Python)	gRPC	Internal calls, shared Protobuf

#2. FastAPI Async Model

#2.1 ASGI Architecture

Python

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_grpc_clients()
    await init_reasoning_engine()
    yield
    await shutdown_grpc_clients()

app = FastAPI(
    title="coomia-dip Intelligence Layer",
    version="1.0.0",
    lifespan=lifespan,
)

#2.2 Route Definitions

Python

from fastapi import APIRouter, Depends, Query, Path
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/v1/reasoning", tags=["reasoning"])

class ReasoningRequest(BaseModel):
    world_id: str = Field(..., description="World ID")
    query: str = Field(..., description="Reasoning query")
    context: dict[str, Any] = Field(default_factory=dict)
    max_depth: int = Field(default=5, ge=1, le=20)

class ReasoningResponse(BaseModel):
    result: dict[str, Any]
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning_chain: list[str]
    execution_time_ms: float

@router.post("/execute", response_model=ReasoningResponse)
async def execute_reasoning(
    request: ReasoningRequest,
    world_context: WorldContext = Depends(get_world_context),
) -> ReasoningResponse:
    result = await reasoning_service.execute(
        world_id=request.world_id,
        query=request.query,
        context=request.context,
        max_depth=request.max_depth,
    )
    return ReasoningResponse(
        result=result.data,
        confidence=result.confidence,
        reasoning_chain=result.chain,
        execution_time_ms=result.duration_ms,
    )

#2.3 Dependency Injection

Python

from fastapi import Depends, Header, HTTPException

async def get_world_context(
    x_world_id: str = Header(..., alias="X-World-Id"),
    x_user_id: str | None = Header(None, alias="X-User-Id"),
) -> WorldContext:
    if not x_world_id:
        raise HTTPException(status_code=400, detail="X-World-Id header required")
    return WorldContext(world_id=x_world_id, user_id=x_user_id)

async def get_reasoning_service(
    world_ctx: WorldContext = Depends(get_world_context),
) -> ReasoningService:
    return ReasoningService(world_context=world_ctx)

#3. gRPC Python Server

#3.1 Protobuf Definition

PROTOBUF

syntax = "proto3";
package onto.intelligence.v1;

service ReasoningService {
    rpc ExecuteReasoning (ReasoningRequest) returns (ReasoningResponse);
    rpc StreamReasoning (ReasoningRequest) returns (stream ReasoningStep);
    rpc BatchReasoning (stream ReasoningRequest) returns (stream ReasoningResponse);
}

message ReasoningRequest {
    string world_id = 1;
    string query = 2;
    map<string, string> context = 3;
    int32 max_depth = 4;
}

message ReasoningResponse {
    map<string, string> result = 1;
    double confidence = 2;
    repeated string reasoning_chain = 3;
    double execution_time_ms = 4;
}

#3.2 gRPC Server Implementation

Python

import grpc
from concurrent import futures

class ReasoningGrpcServicer(reasoning_grpc.ReasoningServiceServicer):

    def __init__(self, reasoning_service: ReasoningService):
        self.service = reasoning_service

    async def ExecuteReasoning(self, request, context):
        metadata = dict(context.invocation_metadata())
        world_id = metadata.get("x-world-id", request.world_id)

        try:
            result = await self.service.execute(
                world_id=world_id,
                query=request.query,
                context=dict(request.context),
                max_depth=request.max_depth,
            )
            return reasoning_pb.ReasoningResponse(
                result=result.data,
                confidence=result.confidence,
                reasoning_chain=result.chain,
                execution_time_ms=result.duration_ms,
            )
        except ValidationError as e:
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, str(e))
        except PermissionError as e:
            await context.abort(grpc.StatusCode.PERMISSION_DENIED, str(e))

    async def StreamReasoning(self, request, context):
        async for step in self.service.stream_execute(
            world_id=request.world_id,
            query=request.query,
        ):
            yield reasoning_pb.ReasoningStep(
                step_number=step.number,
                description=step.description,
                status=step.status,
                intermediate_result=step.data,
            )

#3.3 Dual-Server Startup

Python

import asyncio
import uvicorn

async def start_grpc_server() -> grpc.aio.Server:
    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 30000),
        ],
    )
    reasoning_grpc.add_ReasoningServiceServicer_to_server(
        ReasoningGrpcServicer(reasoning_service), server
    )
    server.add_insecure_port("[::]:50051")
    await server.start()
    return server

async def main():
    grpc_server = await start_grpc_server()
    config = uvicorn.Config(app="main:app", host="0.0.0.0", port=8000, workers=1)
    uvicorn_server = uvicorn.Server(config)
    await asyncio.gather(
        uvicorn_server.serve(),
        grpc_server.wait_for_termination(),
    )

#4. Pydantic v2 and Protobuf Conversion

#4.1 Automatic Conversion Layer

Python

from pydantic import BaseModel
from google.protobuf.message import Message
from google.protobuf.json_format import MessageToDict, ParseDict

class ProtoConverter:
    @staticmethod
    def to_pydantic(proto_msg: Message, pydantic_class: type[BaseModel]) -> BaseModel:
        data = MessageToDict(proto_msg, preserving_proto_field_name=True)
        return pydantic_class.model_validate(data)

    @staticmethod
    def to_proto(pydantic_obj: BaseModel, proto_class: type[Message]) -> Message:
        data = pydantic_obj.model_dump(mode="json")
        return ParseDict(data, proto_class())

#4.2 Shared Model Definition

Python

from pydantic import BaseModel, Field, ConfigDict

class OntologyInstanceModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: str
    object_type: str = Field(alias="objectType")
    object_id: str = Field(alias="objectId")
    world_id: str = Field(alias="worldId")
    properties: dict[str, Any] = Field(default_factory=dict)
    version: int = 0

#5. Unified Error Handling

#5.1 Error Code Mapping

Python

from enum import IntEnum
import grpc

class ErrorCode(IntEnum):
    VALIDATION_ERROR = 1001
    NOT_FOUND = 1002
    PERMISSION_DENIED = 1003
    CONFLICT = 1004
    INTERNAL_ERROR = 1005

ERROR_TO_HTTP = {
    ErrorCode.VALIDATION_ERROR: 400,
    ErrorCode.NOT_FOUND: 404,
    ErrorCode.PERMISSION_DENIED: 403,
    ErrorCode.CONFLICT: 409,
    ErrorCode.INTERNAL_ERROR: 500,
}

ERROR_TO_GRPC = {
    ErrorCode.VALIDATION_ERROR: grpc.StatusCode.INVALID_ARGUMENT,
    ErrorCode.NOT_FOUND: grpc.StatusCode.NOT_FOUND,
    ErrorCode.PERMISSION_DENIED: grpc.StatusCode.PERMISSION_DENIED,
    ErrorCode.CONFLICT: grpc.StatusCode.ALREADY_EXISTS,
    ErrorCode.INTERNAL_ERROR: grpc.StatusCode.INTERNAL,
}

#5.2 Unified Exception Class

Python

class CoomiaDipError(Exception):
    def __init__(self, code: ErrorCode, message: str, details: dict | None = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)

    def to_http_exception(self) -> HTTPException:
        return HTTPException(
            status_code=ERROR_TO_HTTP[self.code],
            detail={"code": self.code, "message": self.message, "details": self.details},
        )

    async def abort_grpc(self, context: grpc.aio.ServicerContext) -> None:
        await context.abort(ERROR_TO_GRPC[self.code], f"[{self.code}] {self.message}")

#6. Middleware and Interceptors

#6.1 FastAPI Middleware

Python

class RequestTracingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-Id", str(uuid4()))
        start_time = time.monotonic()
        token = request_context.set({"request_id": request_id})

        try:
            response = await call_next(request)
            duration = time.monotonic() - start_time
            response.headers["X-Request-Id"] = request_id
            response.headers["X-Response-Time"] = f"{duration:.3f}s"
            return response
        finally:
            request_context.reset(token)

#6.2 gRPC Interceptor

Python

class TracingInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        start_time = time.monotonic()
        method = handler_call_details.method
        try:
            handler = await continuation(handler_call_details)
            duration = time.monotonic() - start_time
            logger.info("gRPC call completed", method=method, duration_ms=duration * 1000)
            return handler
        except Exception as e:
            logger.error("gRPC call failed", method=method, error=str(e))
            raise

#7. Health Checks and Service Discovery

Python

@app.get("/health")
async def health_check():
    checks = {
        "grpc_server": await check_grpc_server(),
        "database": await check_database(),
        "redis": await check_redis(),
    }
    all_healthy = all(checks.values())
    return JSONResponse(
        status_code=200 if all_healthy else 503,
        content={"status": "healthy" if all_healthy else "unhealthy", "checks": checks},
    )

#8. Concurrency Model and Performance

#8.1 asyncio + ThreadPoolExecutor

Python

executor = ThreadPoolExecutor(max_workers=4)

async def cpu_intensive_reasoning(query: str) -> dict:
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, _sync_reasoning, query)
    return result

#8.2 Performance Benchmarks

Code

REST (FastAPI):
  Simple query: 12,000 RPS, P99 = 5ms
  Reasoning:     2,000 RPS, P99 = 45ms

gRPC:
  Simple query: 18,000 RPS, P99 = 2ms
  Reasoning:     3,500 RPS, P99 = 30ms
  Streaming:    50,000 msg/s

gRPC is 50-75% faster than REST (serialization + transport)

#9. Deployment Patterns

YAML

apiVersion: apps/v1
kind: Deployment
metadata:
  name: intelligence-Layer
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: intelligence
          image: coomia-dip/intelligence-Layer:latest
          ports:
            - containerPort: 8000
              name: http
            - containerPort: 50051
              name: grpc
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
          readinessProbe:
            grpc:
              port: 50051

#10. Testing Strategy

Python

def test_execute_reasoning():
    client = TestClient(app)
    response = client.post(
        "/api/v1/reasoning/execute",
        json={"world_id": "test", "query": "optimize route", "max_depth": 3},
        headers={"X-World-Id": "test"},
    )
    assert response.status_code == 200
    assert 0 <= response.json()["confidence"] <= 1

@pytest.mark.asyncio
async def test_grpc_reasoning(grpc_channel):
    stub = reasoning_grpc.ReasoningServiceStub(grpc_channel)
    response = await stub.ExecuteReasoning(
        reasoning_pb.ReasoningRequest(world_id="test", query="optimize route", max_depth=3),
        metadata=[("x-world-id", "test")],
    )
    assert response.confidence > 0

#11. Key Takeaways

Topic	Key Conclusion
Dual protocol	REST for external (browser/SDK), gRPC for internal (microservices)
FastAPI	async/await + Pydantic v2 = high-performance REST
gRPC	grpc.aio async server with streaming support
Model conversion	Pydantic to Protobuf automatic conversion layer
Error handling	Unified error codes mapped to HTTP and gRPC status codes
Concurrency	asyncio for I/O, ThreadPoolExecutor for CPU
Performance	gRPC is 50-75% faster than REST
Deployment	Single process, dual ports; K8s separate probes

“

Next up: S8-16 dives into Apache Arrow Flight SQL, exploring how coomia-dip achieves high-performance columnar data transport.