MCP Server BentoML — service methods, SSE streaming, multimodal input, Bentocloud auth

Calling BentoML service methods from MCP tools

BentoML service methods map directly to HTTP endpoints: a method named summarize becomes POST /summarize. The request body is JSON whose keys match the Python function's parameter names. Unlike OpenAI-style APIs, there is no universal schema — each BentoML service defines its own Pydantic input model. Always fetch /docs on the service to inspect the current schema before writing MCP tool code.

import { z } from 'zod';

const BENTO_URL    = process.env.BENTO_URL    ?? 'http://localhost:3000';
const BENTO_API_KEY = process.env.BENTO_API_KEY;  // undefined for local, required for Bentocloud

// Build authorization headers — Bentocloud needs Bearer token, local needs nothing
function bentoHeaders(extra: Record<string, string> = {}): Record<string, string> {
  const headers: Record<string, string> = {
    'Content-Type': 'application/json',
    ...extra,
  };
  if (BENTO_API_KEY) {
    headers['Authorization'] = `Bearer ${BENTO_API_KEY}`;
  }
  return headers;
}

// Generic BentoML service call — posts to any method endpoint
async function callBento<T = unknown>(
  methodName: string,
  input: Record<string, unknown>,
): Promise<T> {
  const res = await fetch(`${BENTO_URL}/${methodName}`, {
    method:  'POST',
    headers: bentoHeaders(),
    body:    JSON.stringify(input),
    signal:  AbortSignal.timeout(60_000),
  });

  if (!res.ok) {
    const body = await res.text();
    throw new Error(`BentoML ${methodName} failed: HTTP ${res.status} — ${body}`);
  }

  return res.json() as Promise<T>;
}

// MCP tool — text summarization via BentoML
// Assumes a BentoML service with: def summarize(self, text: str, max_length: int = 150) -> str
server.tool(
  'bento_summarize',
  {
    text:       z.string().min(1).max(50_000),
    max_length: z.number().int().min(50).max(500).optional().default(150),
  },
  async ({ text, max_length }) => {
    // Field names must match BentoML service Python parameter names exactly
    const result = await callBento<string>('summarize', { text, max_length });
    return { content: [{ type: 'text', text: result }] };
  }
);

// MCP tool — text classification via BentoML
// Assumes: def classify(self, text: str) -> list[dict] with [{label, score}]
server.tool(
  'bento_classify',
  {
    text: z.string().min(1).max(10_000),
  },
  async ({ text }) => {
    const labels = await callBento<Array<{label: string; score: number}>>('classify', { text });
    const top    = labels.sort((a, b) => b.score - a.score)[0];
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          top_label: top?.label,
          top_score: top?.score,
          all_labels: labels,
        }),
      }],
    };
  }
);

Streaming SSE responses from BentoML

BentoML service methods that return AsyncGenerator[str, None] (Python) stream their output as Server-Sent Events. The HTTP response is Content-Type: text/event-stream with the format data: <chunk>\n\n. You must set Accept: text/event-stream in the request headers to receive the stream — without it, BentoML buffers and returns JSON. The [DONE] sentinel follows the same convention as OpenAI: a final data: [DONE] event signals the stream is complete.

// MCP tool — streaming text generation from a BentoML service
// Assumes: async def generate(self, prompt: str, max_tokens: int) -> AsyncGenerator[str, None]
server.tool(
  'bento_generate_stream',
  {
    prompt:     z.string().min(1).max(8000),
    max_tokens: z.number().int().min(1).max(4096).optional().default(512),
  },
  async ({ prompt, max_tokens }) => {
    const res = await fetch(`${BENTO_URL}/generate`, {
      method: 'POST',
      headers: bentoHeaders({ 'Accept': 'text/event-stream' }),
      body:    JSON.stringify({ prompt, max_tokens }),
      signal:  AbortSignal.timeout(120_000),
    });

    if (!res.ok) {
      const body = await res.text();
      throw new Error(`BentoML generate failed: HTTP ${res.status} — ${body}`);
    }

    // Stream SSE chunks — each line: 'data: \n' or 'data: [DONE]\n'
    if (!res.body) throw new Error('BentoML response has no body');

    const reader  = res.body.getReader();
    const decoder = new TextDecoder();
    let   buffer  = '';
    let   fullText = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });

      // SSE lines end with '\n\n' — split on double newline to get events
      const events = buffer.split('\n\n');
      buffer = events.pop() ?? '';  // keep partial last event in buffer

      for (const event of events) {
        // Each event block may have multiple 'data: ...' lines
        for (const line of event.split('\n')) {
          if (!line.startsWith('data: ')) continue;
          const data = line.slice(6);  // strip 'data: ' prefix

          if (data === '[DONE]') break;  // BentoML stream complete

          fullText += data;
        }
      }
    }

    return { content: [{ type: 'text', text: fullText }] };
  }
);

Multimodal file input — images and audio

BentoML services that accept images, audio, or other binary files use multipart/form-data rather than JSON. The field name in the form data must match the Python parameter name annotated with bentoml.Image, bentoml.File, or bentoml.Audio. Do not send binary content as base64 in a JSON body — BentoML will reject it with a 422 validation error for file-typed parameters.

import { z } from 'zod';

// MCP tool — image captioning via multimodal BentoML service
// Assumes: def caption(self, image: bentoml.Image) -> str
server.tool(
  'bento_caption_image',
  {
    image_url: z.string().url(),  // agent provides a URL to an image
    prompt:    z.string().optional().default('Describe this image in detail.'),
  },
  async ({ image_url, prompt }) => {
    // Download the image bytes from the URL
    const imageRes = await fetch(image_url, { signal: AbortSignal.timeout(15_000) });
    if (!imageRes.ok) throw new Error(`Failed to fetch image: ${imageRes.status}`);

    const imageBuffer = await imageRes.arrayBuffer();
    const contentType = imageRes.headers.get('content-type') ?? 'image/jpeg';
    const ext         = contentType.split('/')[1]?.split(';')[0] ?? 'jpg';

    // BentoML multimodal services use multipart/form-data
    const form = new FormData();
    // Field name must match the Python parameter name exactly
    form.append('image', new Blob([imageBuffer], { type: contentType }), `image.${ext}`);
    form.append('prompt', prompt);  // additional parameters go alongside the file

    const headers: Record<string, string> = {};
    if (BENTO_API_KEY) headers['Authorization'] = `Bearer ${BENTO_API_KEY}`;
    // Do NOT set Content-Type manually — FormData sets it with the boundary

    const res = await fetch(`${BENTO_URL}/caption`, {
      method:  'POST',
      headers,
      body:    form,
      signal:  AbortSignal.timeout(60_000),
    });

    if (!res.ok) {
      const body = await res.text();
      throw new Error(`BentoML caption failed: HTTP ${res.status} — ${body}`);
    }

    const caption = await res.text();  // BentoML returns str as plain text
    return { content: [{ type: 'text', text: caption }] };
  }
);

Health probe and readiness check

BentoML services expose GET /healthz and GET /readyz. The /healthz endpoint returns 200 once the service process is running. The /readyz endpoint returns 200 only after model weights are fully loaded into memory — use /readyz for startup probes in MCP server initialization to avoid tool calls that hit a service that is still loading its model.

// MCP resource — BentoML readiness check
server.resource('bento_health', 'bento://health', async () => {
  try {
    // Use /readyz — model fully loaded (not just process alive)
    const res = await fetch(`${BENTO_URL}/readyz`, {
      headers: BENTO_API_KEY ? { Authorization: `Bearer ${BENTO_API_KEY}` } : {},
      signal:  AbortSignal.timeout(5_000),
    });

    const isReady = res.status === 200;
    return {
      contents: [{
        uri: 'bento://health',
        text: JSON.stringify({
          status:      isReady ? 'ok' : 'loading',
          http_status: res.status,
          endpoint:    BENTO_URL,
          checked_at:  new Date().toISOString(),
        }),
      }],
    };
  } catch (err: any) {
    return {
      contents: [{
        uri: 'bento://health',
        text: JSON.stringify({ status: 'unreachable', message: err.message }),
      }],
    };
  }
});

// Startup wait — poll /readyz until model is loaded
async function waitForBento(maxWaitMs = 180_000): Promise<void> {
  const start   = Date.now();
  const headers = BENTO_API_KEY ? { Authorization: `Bearer ${BENTO_API_KEY}` } : {};

  while (Date.now() - start < maxWaitMs) {
    try {
      const res = await fetch(`${BENTO_URL}/readyz`, {
        headers,
        signal: AbortSignal.timeout(5_000),
      });
      if (res.status === 200) return;
    } catch {}

    await new Promise(r => setTimeout(r, 5_000));
  }

  throw new Error(`BentoML service at ${BENTO_URL} did not become ready within ${maxWaitMs / 1000}s`);
}

Bentocloud deployments also expose a metrics endpoint at /metrics (Prometheus format) that includes bentoml_service_request_total, bentoml_service_request_duration_seconds, and per-runner utilization metrics. Scrape these in your AliveMCP uptime monitor to track not just whether the service is alive but whether its per-request latency is drifting.