Implementation:Bentoml BentoML IO Descriptor Text
| Knowledge Sources | |
|---|---|
| Domains | IO Descriptors, Text Processing, HTTP, gRPC, OpenAPI |
| Last Updated | 2026-02-13 15:00 GMT |
Overview
Implements the Text IO descriptor for BentoML, handling plain text and server-sent event (SSE) streams as input/output for inference API endpoints.
Description
The Text class is a concrete IODescriptor[str] subclass registered with descriptor ID bentoml.io.Text and proto field text. It handles string data for both HTTP and gRPC protocols.
The constructor accepts an optional content_type parameter that can be either "text/plain" (default) or "text/event-stream" for SSE streaming responses.
For HTTP:
- from_http_request() reads the request body and decodes it as UTF-8 to produce a string.
- to_http_response() wraps the output in a StreamingResponse. If the output is a string, it is wrapped in a single-item iterator; if it is an AsyncGenerator[str, None], it streams directly. When a ServiceContext is provided, response headers, status code, and cookies are applied.
For gRPC:
- from_proto() accepts either raw bytes or a google.protobuf.wrappers_pb2.StringValue and returns a string.
- to_proto() wraps the string in a StringValue protobuf message.
OpenAPI integration provides a string schema type with the configured MIME type for both request body and response specifications.
The _from_sample() method converts bytes to string (decoding UTF-8) if needed, enabling Text.from_sample() usage.
Usage
Use Text() when your BentoML service API accepts or returns plain text strings. Use Text(content_type="text/event-stream") for streaming server-sent event responses.
Code Reference
Source Location
- Repository: Bentoml_BentoML
- File: src/bentoml/_internal/io_descriptors/text.py
- Lines: 1-187
Signature
class Text(
IODescriptor[str],
descriptor_id="bentoml.io.Text",
proto_fields=("text",),
):
def __init__(
self,
content_type: t.Literal["text/plain", "text/event-stream"] = "text/plain",
): ...
def _from_sample(self, sample: str | bytes) -> str: ...
def input_type(self) -> t.Type[str]: ...
def to_spec(self) -> dict[str, t.Any]: ...
@classmethod
def from_spec(cls, spec: dict[str, t.Any]) -> Self: ...
def openapi_schema(self) -> Schema: ...
def openapi_components(self) -> dict[str, t.Any] | None: ...
def openapi_example(self): ...
def openapi_request_body(self) -> dict[str, t.Any]: ...
def openapi_responses(self) -> OpenAPIResponse: ...
async def from_http_request(self, request: Request) -> str: ...
async def to_http_response(
self, obj: str | t.AsyncGenerator[str, None], ctx: Context | None = None,
) -> StreamingResponse: ...
async def from_proto(self, field: wrappers_pb2.StringValue | bytes) -> str: ...
async def to_proto(self, obj: str) -> wrappers_pb2.StringValue: ...
Import
from bentoml.io import Text
# Or direct import
from bentoml._internal.io_descriptors.text import Text
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| content_type | Literal["text/plain", "text/event-stream"] | No (default "text/plain") | MIME type for the text content |
| request | starlette.requests.Request | Yes (from_http_request) | HTTP request with text body |
| field | StringValue or bytes | Yes (from_proto) | gRPC proto field containing text data |
| sample | str or bytes | Yes (from_sample) | Sample text for creating the descriptor |
Outputs
| Name | Type | Description |
|---|---|---|
| str (from_http_request) | str | Decoded UTF-8 string from the request body |
| StreamingResponse (to_http_response) | StreamingResponse | Starlette streaming response with text content |
| str (from_proto) | str | Decoded string from gRPC proto field |
| StringValue (to_proto) | wrappers_pb2.StringValue | Protobuf StringValue wrapping the output text |
Usage Examples
import bentoml
from bentoml.io import Text
runner = bentoml.models.get("gpt2:latest").to_runner()
svc = bentoml.legacy.Service("gpt2-generation", runners=[runner])
# Simple text-in, text-out API
@svc.api(input=Text(), output=Text())
def predict(text: str) -> str:
res = runner.run(text)
return res["generated_text"]
# Streaming SSE response
@svc.api(input=Text(), output=Text(content_type="text/event-stream"))
async def stream_predict(text: str) -> t.AsyncGenerator[str, None]:
async for token in runner.async_run(text):
yield token
# Creating from sample
text_descriptor = Text.from_sample("Bento box is")
# Using with curl:
# curl -X POST -H "Content-Type: text/plain" \
# --data "Not for nothing did Orin say that people outdoors." \
# http://0.0.0.0:3000/predict