🔌 Epic: gRPC-to-MCP Protocol Translation

Goal

Enable seamless integration of gRPC microservices into the MCP Gateway ecosystem, allowing AI agents and LLM applications to discover and invoke gRPC services through the Model Context Protocol. This bridges the gap between traditional microservices architectures and the emerging MCP ecosystem, unlocking enterprise gRPC services for AI agent workflows.

Why Now?

Many organizations have invested heavily in gRPC-based microservices for performance, type safety, and language-agnostic APIs. However, these services are currently inaccessible to MCP clients and AI agents. By implementing automatic gRPC-to-MCP translation, we:

1. Expand MCP's reach into enterprise environments with existing gRPC infrastructure
2. Eliminate manual wrapper development by auto-discovering services via Server Reflection
3. Enable AI agents to interact with high-performance backend services without custom integration code
4. Provide protocol translation between Protobuf binary format and MCP's JSON-RPC

This feature positions MCP Gateway as a universal protocol bridge, not just an MCP registry.

📖 User Stories

Given a gRPC server at "payments.example.com:50051" with Server Reflection enabled
When I register it via "POST /grpc" API or CLI with target address
Then the gateway should:
  - Connect to the gRPC server
  - Use Server Reflection to list all services
  - Extract method signatures and message types
  - Store service descriptors in the database
  - Mark the service as "reachable" if successful
  - Create MCP tool definitions for each gRPC method

Given a registered gRPC service "PaymentService" with method "ProcessPayment"
When an MCP client sends a tool invocation:
  {
    "tool": "PaymentService.ProcessPayment",
    "arguments": {
      "amount": 99.99,
      "currency": "USD",
      "customer_id": "cust_123"
    }
  }
Then the gateway should:
  - Convert JSON arguments to Protobuf PaymentRequest message
  - Invoke PaymentService.ProcessPayment via gRPC
  - Receive Protobuf PaymentResponse
  - Convert response back to JSON
  - Return MCP tool result with JSON payload

Given a gRPC service requiring mTLS authentication
When I register the service with:
  - tls_enabled: true
  - tls_cert_path: "/path/to/client.crt"
  - tls_key_path: "/path/to/client.key"
Then the gateway should:
  - Load the client certificate and private key
  - Create a secure gRPC channel with ssl_channel_credentials
  - Establish mTLS connection with the server
  - Perform reflection and method invocation over secure channel

Given I access the MCP Gateway Admin UI
When I navigate to the "gRPC Services" tab
Then I should see:
  - List of registered gRPC services with status indicators
  - "Add Service" button to register new endpoints
  - Form with fields: name, target, description, TLS settings, metadata
  - "Test Connection" button to verify reachability
  - "Discover Services" button to trigger reflection
  - Service details showing discovered methods and schemas

Given a gRPC service with server-streaming method "StreamLogs"
When I invoke the method via MCP Gateway
Then the gateway should:
  - Create a unary_stream gRPC call
  - Yield JSON responses as they arrive from the stream
  - Support SSE transport for incremental delivery to MCP client
  - Handle stream completion and errors gracefully

🏗 Architecture

System Architecture Diagram

graph TB
    subgraph "MCP Clients"
        A1[AI Agent]
        A2[LLM App]
    end

    subgraph "MCP Gateway"
        B1[REST API /grpc]
        B2[Admin UI]
        B3[GrpcService Layer]
        B4[GrpcEndpoint]
        B5[Protocol Translator]
        B6[Descriptor Pool]
    end

    subgraph "Backend Services"
        C1[Payment Service :50051]
        C2[User Service :50052]
        C3[Inventory Service :50053]
    end

    A1 -->|MCP JSON-RPC| B1
    A2 -->|MCP SSE| B1
    B2 -->|HTMX| B3
    B1 --> B3
    B3 --> B4
    B4 --> B5
    B5 --> B6
    B4 -->|gRPC Reflection| C1
    B4 -->|gRPC Unary Call| C2
    B4 -->|gRPC Streaming| C3
    B6 -.Protobuf Schemas.-> B5

Service Discovery Flow

sequenceDiagram
    participant Admin as Admin UI
    participant API as REST API
    participant Service as GrpcService
    participant Endpoint as GrpcEndpoint
    participant gRPC as gRPC Server
    participant DB as Database

    Admin->>API: POST /grpc {target, tls_enabled}
    API->>Service: register_service()
    Service->>DB: Create DbGrpcService
    Service->>Endpoint: GrpcEndpoint(target, tls_enabled)
    Endpoint->>gRPC: Connect (secure/insecure channel)
    Endpoint->>gRPC: ServerReflectionInfo (list_services)
    gRPC-->>Endpoint: [service names]

    loop For each service
        Endpoint->>gRPC: ServerReflectionInfo (file_containing_symbol)
        gRPC-->>Endpoint: FileDescriptorProto
        Endpoint->>Endpoint: Parse service/method descriptors
        Endpoint->>Endpoint: Add to descriptor pool
    end

    Endpoint-->>Service: discovered_services, service_count, method_count
    Service->>DB: Update DbGrpcService
    Service-->>API: GrpcServiceRead
    API-->>Admin: 201 Created

Method Invocation Flow

sequenceDiagram
    participant Client as MCP Client
    participant Gateway as MCP Gateway
    participant Service as GrpcService
    participant Endpoint as GrpcEndpoint
    participant Pool as Descriptor Pool
    participant gRPC as gRPC Server

    Client->>Gateway: POST /tools/invoke {"tool": "PaymentService.ProcessPayment", "arguments": {...}}
    Gateway->>Service: invoke_method(service_id, method_name, request_data)
    Service->>Endpoint: GrpcEndpoint(target, tls_enabled)
    Service->>Endpoint: invoke(service, method, request_data)

    Endpoint->>Pool: FindMessageTypeByName("PaymentRequest")
    Pool-->>Endpoint: request_descriptor
    Endpoint->>Endpoint: json_format.ParseDict(request_data, request_msg)

    Endpoint->>gRPC: unary_unary("/PaymentService/ProcessPayment", request_msg)
    gRPC-->>Endpoint: response_msg (Protobuf)

    Endpoint->>Endpoint: json_format.MessageToDict(response_msg)
    Endpoint-->>Service: response_dict (JSON)
    Service-->>Gateway: response_dict
    Gateway-->>Client: MCP ToolResult

🗂 Database Schema

GrpcService Table

class GrpcService(Base):
    """Database model for gRPC service registrations."""

    __tablename__ = "grpc_services"

    # Identity
    id: str = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)
    name: str = Column(String, unique=True, nullable=False, index=True)
    slug: str = Column(String, unique=True, nullable=False, index=True)

    # Connection
    target: str = Column(String, nullable=False)  # "host:port"
    reflection_enabled: bool = Column(Boolean, default=True)

    # TLS Configuration
    tls_enabled: bool = Column(Boolean, default=False)
    tls_cert_path: Optional[str] = Column(String, nullable=True)
    tls_key_path: Optional[str] = Column(String, nullable=True)

    # Authentication
    grpc_metadata: Dict[str, str] = Column(JSON, default=dict)  # Custom headers

    # Discovery Results
    discovered_services: Dict[str, Any] = Column(JSON, default=dict)
    service_count: int = Column(Integer, default=0)
    service_count: int = Column(Integer, default=0)
    method_count: int = Column(Integer, default=0)
    last_reflection: Optional[datetime] = Column(DateTime, nullable=True)

    # Status
    enabled: bool = Column(Boolean, default=True)
    reachable: bool = Column(Boolean, default=False)

    # Metadata
    description: Optional[str] = Column(String, nullable=True)
    tags: List[str] = Column(JSON, default=list)

    # Team/Access Control
    team_id: Optional[str] = Column(String, ForeignKey("teams.id"), nullable=True)
    owner_email: Optional[str] = Column(String, nullable=True)
    visibility: str = Column(String, default="public")  # private, team, public

⚙️ Configuration

Environment Variables

# Feature Flag
MCPGATEWAY_GRPC_ENABLED=true

# Discovery Settings
GRPC_REFLECTION_TIMEOUT=30  # seconds
GRPC_CONNECTION_TIMEOUT=10   # seconds

# Retry Configuration
GRPC_MAX_RETRIES=3
GRPC_RETRY_DELAY=1  # seconds

# Performance
GRPC_POOL_SIZE=10           # Connection pool per service
GRPC_MAX_MESSAGE_SIZE=4194304  # 4MB

Service Configuration Example

@dataclass
class GrpcServiceConfig:
    """Configuration for a gRPC service registration."""

    name: str
    target: str  # "payments.example.com:50051"
    description: Optional[str] = None

    # Connection
    reflection_enabled: bool = True
    tls_enabled: bool = False
    tls_cert_path: Optional[str] = None
    tls_key_path: Optional[str] = None

    # Authentication
    grpc_metadata: Dict[str, str] = field(default_factory=dict)

    # Access Control
    team_id: Optional[str] = None
    owner_email: Optional[str] = None
    visibility: str = "public"

    # Tags
    tags: List[str] = field(default_factory=list)

📋 Implementation Tasks

Phase 1: Core Infrastructure ✅ (Assumed Not Implemented)

• Backend Models & Database

  • Create GrpcService SQLAlchemy model in db.py
  • Add Alembic migration for grpc_services table
  • Implement grpc_metadata field (avoid reserved metadata name)
  • Add team-based access control columns

• Pydantic Schemas

  • Create GrpcServiceCreate schema with validation
  • Create GrpcServiceUpdate schema
  • Create GrpcServiceRead schema
  • Add field validators for target (host:port format)
  • Implement description sanitization

• Configuration

  • Add MCPGATEWAY_GRPC_ENABLED to config.py
  • Document gRPC-related env vars in .env.example
  • Update Settings class with gRPC flags

Phase 2: Service Discovery & Reflection ✅

• GrpcEndpoint Class (translate_grpc.py)

  • Implement __init__() with channel creation logic
  • Add TLS channel creation with certificate loading
  • Implement start() method for connection establishment
  • Create _discover_services() using Server Reflection
  • Implement _discover_service_details() for FileDescriptorProto parsing
  • Build descriptor pool and message factory
  • Parse service/method metadata from Protobuf descriptors
  • Handle reflection service exclusion
  • Implement close() for channel cleanup

• Protocol Translation

  • Implement protobuf_to_json_schema() with 18 type mappings
  • Create _protobuf_field_to_json_schema() helper
  • Handle repeated fields (arrays)
  • Handle nested messages (recursive schemas)
  • Implement GrpcToMcpTranslator class
  • Create grpc_service_to_mcp_server() converter
  • Create grpc_methods_to_mcp_tools() converter

Phase 3: Method Invocation ✅

• Unary RPC Support

  • Implement invoke() method in GrpcEndpoint
  • Add JSON-to-Protobuf conversion using json_format.ParseDict()
  • Create dynamic gRPC channel.unary_unary() calls
  • Add Protobuf-to-JSON conversion using json_format.MessageToDict()
  • Handle service/method validation
  • Implement proper error handling and exceptions

• Server-Streaming RPC Support

  • Implement invoke_streaming() async generator
  • Create channel.unary_stream() calls
  • Validate server_streaming flag
  • Yield JSON responses as they arrive
  • Handle grpc.RpcError exceptions
  • Support stream cancellation

Phase 4: Service Layer ✅

• GrpcService Class (services/grpc_service.py)

  • Implement register_service() method
  • Add _perform_reflection() for discovery
  • Create list_services() with team filtering
  • Implement get_service() by ID
  • Add update_service() method
  • Implement delete_service() method
  • Create invoke_method() for tool invocation
  • Add get_service_methods() for method listing
  • Integrate with TeamManagementService

• Error Handling

  • Define GrpcServiceError exception
  • Define GrpcServiceNotFoundError exception
  • Handle connection failures gracefully
  • Handle TLS certificate errors
  • Log detailed error messages

Phase 5: REST API & Admin UI ✅

• REST Endpoints (admin.py)

  • POST /grpc - Register new gRPC service
  • GET /grpc - List all gRPC services
  • GET /grpc/{id} - Get service details
  • PUT /grpc/{id} - Update service configuration
  • DELETE /grpc/{id} - Delete service
  • POST /grpc/{id}/discover - Trigger reflection
  • GET /grpc/{id}/methods - List methods
  • POST /grpc/{id}/invoke - Test method invocation

• Admin UI Tab

  • Add "gRPC Services" tab to admin.html
  • Create service registration form
  • Implement service list with status indicators
  • Add "Test Connection" functionality
  • Create service details view
  • Add inline enable/disable toggle
  • Implement JavaScript handlers in admin.js
  • Add error handling and notifications

Phase 6: CLI Integration ✅

• Translate CLI (translate.py)
  • Add --grpc argument for gRPC target
  • Implement expose_grpc_via_sse() function
  • Create standalone gRPC-to-SSE server
  • Add TLS arguments (--tls-cert, --tls-key)
  • Add metadata header arguments
  • Implement graceful shutdown handling

Phase 7: Documentation ✅

• User Documentation

  • Create docs/docs/using/grpc-services.md guide
  • Update docs/docs/using/mcpgateway-translate.md with gRPC examples
  • Add gRPC section to docs/docs/overview/features.md
  • Document TLS/mTLS configuration
  • Provide example gRPC service integration

• Configuration Documentation

  • Update .env.example with gRPC variables
  • Document MCPGATEWAY_GRPC_ENABLED flag
  • Add gRPC mentions to README.md

Phase 8: Testing ✅

• Unit Tests - translate_grpc.py

  • Test GrpcEndpoint initialization (insecure, TLS, metadata)
  • Test channel creation logic
  • Test _discover_services() with mocked reflection
  • Test _discover_service_details() descriptor parsing
  • Test invoke() method with JSON↔Protobuf conversion
  • Test invoke_streaming() async generator
  • Test error handling (service not found, method not found)
  • Test GrpcToMcpTranslator conversions
  • Test protobuf_to_json_schema() type mappings

• Unit Tests - grpc_service.py

  • Test service registration flow
  • Test reflection performance
  • Test list/get/update/delete operations
  • Test team-based filtering
  • Test method invocation
  • Test error scenarios (unreachable service, invalid certificates)
  • Test enable/disable functionality

• Integration Tests

  • Test end-to-end service registration via API
  • Test MCP tool invocation with real gRPC server
  • Test Admin UI workflows
  • Test CLI standalone mode

Phase 9: Quality & Polish ✅

• Code Quality

  • Fix all flake8 issues
  • Add comprehensive docstrings
  • Update type hints
  • Run make autoflake isort black
  • Pass make verify checks

• Performance

  • Add connection pooling for gRPC channels
  • Implement descriptor pool caching
  • Optimize reflection frequency
  • Add monitoring metrics

✅ Success Criteria

• Discovery: Can auto-discover all services from a gRPC server via Server Reflection
• Translation: Can convert JSON requests to Protobuf and invoke gRPC methods
• Streaming: Can handle server-streaming RPCs with incremental JSON responses
• Security: Can establish TLS/mTLS connections with custom certificates
• API: Complete REST API for CRUD operations on gRPC services
• UI: Admin UI tab for visual management and testing
• CLI: Standalone CLI mode to expose gRPC services via SSE
• Tests: 40+ unit tests with 60%+ coverage for gRPC modules
• Docs: Complete user guide with examples and configuration
• Quality: All code passes flake8, has proper docstrings, and builds successfully

📝 Additional Notes

🔹 Automatic Schema Discovery: Unlike REST APIs that require OpenAPI specs, gRPC services expose their schemas via Server Reflection Protocol, enabling zero-configuration integration.

🔹 Binary Performance: Protobuf binary encoding provides 3-10x smaller payloads and 20-100x faster serialization compared to JSON, ideal for high-throughput AI agent workflows.

🔹 Type Safety: gRPC's strong typing (via .proto files) prevents runtime schema mismatches that plague JSON-based integrations.

🔹 Enterprise Adoption: Most major cloud platforms (Google Cloud, AWS, Azure) use gRPC for internal services. This feature unlocks those ecosystems for MCP clients.

🔹 Streaming Support: Server-streaming RPCs enable real-time data feeds (logs, metrics, events) to AI agents without polling.

🔹 Future Extensions:

• Client-streaming RPC support
• Bidirectional streaming support
• gRPC-Web integration for browser clients
• Automatic Protobuf schema publication

🏁 Definition of Done

• All tasks in all phases completed
• 40+ unit tests passing with 60%+ coverage
• Integration tests verify end-to-end flows
• Documentation complete (user guide, API docs, configuration)
• Admin UI fully functional for service management
• CLI supports standalone gRPC-to-SSE mode
• Code passes all quality checks (flake8, black, isort)
• Build verification passes (make verify)
• Feature demonstrated with real gRPC service
• Team review completed and approved