Argumentation Microservices for Python

Generated protobuf, gRPC, and Connect code of the arg-services Protobuf definitions for building complex argumentation machines. Documentation can be found at the Buf Schema Registry.

Build Locally

buf generate && uv build

Runtimes

All messages (*_pb2.py) use the official protobuf runtime. Install the extra for the RPC runtime you want to use, both share the same messages:

Validation

arg_services.validation checks requests independently of the runtime and raises a ValidationError. Add the ValidationInterceptor of your runtime to report it as INVALID_ARGUMENT, e.g., serve(..., interceptors=[ValidationInterceptor()]). To pass the traceback of unhandled exceptions to gRPC clients, also add arg_services.grpcio.ExceptionInterceptor(). For asyncio servicers, use arg_services.grpcio.serve_async with AsyncValidationInterceptor() and AsyncExceptionInterceptor() instead.

from arg_services.validation import require_all

require_all(["texts", "config.language"], request)
require_all(["text"], *request.items)

Connect

Connect services are plain ASGI/WSGI applications that also speak gRPC, so existing gRPC clients keep working. gRPC clients and reflection need a server with HTTP/2 trailers such as pyvoy, Connect clients work with any server such as uvicorn or gunicorn. Since connectrpc-grpcreflect only supports protobuf-py, use arg_services.connectrpc.reflection_app to serve gRPC reflection.

from arg_services.connectrpc import ValidationInterceptor, reflection_app
from arg_services.nlp.v1 import nlp_pb2
from arg_services.nlp.v1.nlp_connect import NlpService, NlpServiceASGIApplication
from arg_services.validation import require_all
from connectrpc.request import RequestContext
from starlette.applications import Starlette
from starlette.routing import Mount


class Nlp(NlpService):
    async def vectors(
        self, request: nlp_pb2.VectorsRequest, ctx: RequestContext
    ) -> nlp_pb2.VectorsResponse:
        require_all(["texts"], request)

        return nlp_pb2.VectorsResponse()


nlp_app = NlpServiceASGIApplication(Nlp(), interceptors=[ValidationInterceptor()])
reflection = reflection_app(nlp_app)
app = Starlette(
    routes=[Mount(nlp_app.path, nlp_app), Mount(reflection.path, reflection)]
)