Modular API

from baml_client import b
async def run():
  # HTTP request + LLM response parsing.
  resume = await b.ExtractResume("John Doe | Software Engineer | BSc in CS")
  print(resume)

import requests
# requests is not async so for simplicity we'll use the sync client.
from baml_client.sync_client import b
def run():
  # Get the HTTP request object.
  req = b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  # Send the HTTP request.
  res = requests.post(url=req.url, headers=req.headers, json=req.body.json())
  # Parse the LLM response.
  parsed = b.parse.ExtractResume(res.json()["choices"][0]["message"]["content"])
  # Fully parsed Resume type.
  print(parsed)

res = requests.post(url=req.url, headers=req.headers, data=req.body.raw())

from openai import AsyncOpenAI
from baml_client import b
async def run():
  # Initialize the OpenAI client.
  client = AsyncOpenAI()
  # Get the HTTP request object.
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  # Use the openai library to send the request.
  res = await client.chat.completions.create(**req.body.json())
  # Parse the LLM response.
  parsed = b.parse.ExtractResume(res.choices[0].message.content)
  # Fully parsed Resume type.
  print(parsed)

from openai import AsyncOpenAI
from openai.types.responses import Response
from baml_client import b
import typing
async def run():
  # Initialize the OpenAI client.
  client = AsyncOpenAI()
  # Get the HTTP request object from a function using openai-responses provider.
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  # Use the openai responses API endpoint.
  res = typing.cast(Response, await client.responses.create(**req.body.json()))
  # Parse the LLM response from the responses API.
  parsed = b.parse.ExtractResume(res.output_text)
  # Fully parsed Resume type.
  print(parsed)

import anthropic
from baml_client import b
async def run():
  # Initialize the Anthropic client.
  client = anthropic.AsyncAnthropic()
  # Get the HTTP request object.
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  # Use the anthropic library to send the request.
  res = await client.messages.create(**req.body.json())
  # Parse the LLM response.
  parsed = b.parse.ExtractResume(res.content[0].text)
  # Fully parsed Resume type.
  print(parsed)

from google import genai
from baml_client import b
async def run():
  # Initialize the Gemini client.
  client = genai.Client()
  # Get the HTTP request object.
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  # Get the request body.
  body = req.body.json()
  # Use the gemini library to send the request.
  res = await client.aio.models.generate_content(
    model="gemini-3.5-flash",
    contents=body["contents"],
    config={
      "safety_settings": [body["safetySettings"]] # REST API uses camelCase
    }
  )
  # Parse the LLM response.
  parsed = b.parse.ExtractResume(res.text)
  # Fully parsed Resume type.
  print(parsed)

import asyncio
import json
import os
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
import boto3
from baml_client import b
from urllib.parse import urlsplit
async def run():
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  body = req.body.json()
  # Optional: append your own messages before signing.
  body["messages"].append({
    "role": "system",
    "content": [{"text": "You must respond in JSON."}],
  })
  body_string = json.dumps(body)
  body_bytes = body_string.encode("utf-8")
  session = boto3.Session()
  credentials = session.get_credentials().get_frozen_credentials()
  region = (
    req.client_details.options.get("region")
    or os.environ.get("AWS_REGION")
    or os.environ.get("AWS_DEFAULT_REGION")
    or session.region_name
    or "us-east-1"
  )
  url = urlsplit(req.url)
  base_headers = {
    key: value
    for key, value in dict(req.headers).items()
    if value is not None
  }
  headers = {
    **base_headers,
    "content-type": "application/json",
    "accept": "application/json",
    "host": url.netloc,
  }
  aws_request = AWSRequest(
    method=req.method,
    url=req.url,
    data=body_bytes,
    headers=headers,
  )
  SigV4Auth(credentials, "bedrock", region).add_auth(aws_request)
  async with httpx.AsyncClient() as client:
    response = await client.post(
      req.url,
      headers={key: str(value) for key, value in aws_request.headers.items()},
      content=body_bytes,
    )
    if not response.is_success:
      raise RuntimeError(
        f"Bedrock request failed: {response.status_code} {response.text}"
      )
  payload = response.json()
  message = payload["output"]["message"]["content"][0]["text"]
  parsed = b.parse.ExtractResume(message)
  print(parsed)
asyncio.run(run())

import typing
from openai import AsyncOpenAI, AsyncStream
from openai.types.chat import ChatCompletionChunk
from baml_client import b
async def run():
  client = AsyncOpenAI()
  req = await b.stream_request.ExtractResume("John Doe | Software Engineer | BSc in CS")
  stream = typing.cast(
    AsyncStream[ChatCompletionChunk],
    await client.chat.completions.create(**req.body.json())
  )
  llm_response: list[str] = []
  async for chunk in stream:
    if len(chunk.choices) > 0 and chunk.choices[0].delta.content is not None:
      llm_response.append(chunk.choices[0].delta.content)
      # You can parse the partial responses as they come in.
      print(b.parse_stream.ExtractResume("".join(llm_response)))

import asyncio
import json
from openai import AsyncOpenAI
from baml_py import HTTPRequest as BamlHttpRequest
from baml_client import b
from baml_client import types
async def run():
  client = AsyncOpenAI()
  # Build the batch requests with BAML.
  john_req, jane_req = await asyncio.gather(
    b.request.ExtractResume("John Doe | Software Engineer | BSc in CS"),
    b.request.ExtractResume("Jane Smith | Data Scientist | PhD in Statistics"),
  )
  # Build the JSONL content.
  jsonl = to_openai_jsonl(john_req) + to_openai_jsonl(jane_req)
  # Create the batch input file.
  batch_input_file = await client.files.create(
    file=jsonl.encode("utf-8"),
    purpose="batch",
  )
  # Create the batch.
  batch = await client.batches.create(
    input_file_id=batch_input_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={
      "description": "BAML Modular API Python Batch Example"
    },
  )
  # Wait for the batch to complete (exponential backoff).
  backoff = 2
  attempts = 0
  max_attempts = 5
  while True:
    batch = await client.batches.retrieve(batch.id)
    attempts += 1
    if batch.status == "completed":
        break
    if attempts >= max_attempts:
      try:
        await client.batches.cancel(batch.id)
      finally:
        raise Exception("Batch failed to complete in time")
    await asyncio.sleep(backoff)
    back_off *= 2
  # Retrieve the batch output file.
  output = await client.files.content(batch.output_file_id)
  # You can match the batch results using the BAML request IDs.
  expected = {
    john_req.id: types.Resume(
      name="John Doe",
      experience=["Software Engineer"],
      education=["BSc in CS"]
    ),
    jane_req.id: types.Resume(
      name="Jane Smith",
      experience=["Data Scientist"],
      education=["PhD in Statistics"]
    ),
  }
  resumes = {}
  for line in output.text.splitlines():
    result = json.loads(line)
    llm_response = result["response"]["body"]["choices"][0]["message"]["content"]
    parsed = b.parse.ExtractResume(llm_response)
    resumes[result["custom_id"]] = parsed
  print(resumes)
  # Should be equal.
  assert resumes == expected
def to_openai_jsonl(req: BamlHttpRequest) -> str:
  """ Helper that converts a BAML HTTP request to OpenAI JSONL format. """
  line = json.dumps({
    "custom_id": req.id, # Important for matching the batch results.
    "method": "POST",
    "url": "/v1/chat/completions",
    "body": req.body.json(),
  })
  return f"{line}\n"