Modular API

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

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

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

1
from openai import AsyncOpenAI
2
from baml_client import b
3
4
async def run():
5
  # Initialize the OpenAI client.
6
  client = AsyncOpenAI()
7
8
  # Get the HTTP request object.
9
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
10
11
  # Use the openai library to send the request.
12
  res = await client.chat.completions.create(**req.body.json())
13
14
  # Parse the LLM response.
15
  parsed = b.parse.ExtractResume(res.choices[0].message.content)
16
17
  # Fully parsed Resume type.
18
  print(parsed)

1
from openai import AsyncOpenAI
2
from openai.types.responses import Response
3
from baml_client import b
4
import typing
5
6
async def run():
7
  # Initialize the OpenAI client.
8
  client = AsyncOpenAI()
9
10
  # Get the HTTP request object from a function using openai-responses provider.
11
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
12
13
  # Use the openai responses API endpoint.
14
  res = typing.cast(Response, await client.responses.create(**req.body.json()))
15
16
  # Parse the LLM response from the responses API.
17
  parsed = b.parse.ExtractResume(res.output_text)
18
19
  # Fully parsed Resume type.
20
  print(parsed)

1
import anthropic
2
from baml_client import b
3
4
async def run():
5
  # Initialize the Anthropic client.
6
  client = anthropic.AsyncAnthropic()
7
8
  # Get the HTTP request object.
9
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
10
11
  # Use the anthropic library to send the request.
12
  res = await client.messages.create(**req.body.json())
13
14
  # Parse the LLM response.
15
  parsed = b.parse.ExtractResume(res.content[0].text)
16
17
  # Fully parsed Resume type.
18
  print(parsed)

1
from google import genai
2
from baml_client import b
3
4
async def run():
5
  # Initialize the Gemini client.
6
  client = genai.Client()
7
8
  # Get the HTTP request object.
9
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
10
11
  # Get the request body.
12
  body = req.body.json()
13
14
  # Use the gemini library to send the request.
15
  res = await client.aio.models.generate_content(
16
    model="gemini-3.5-flash",
17
    contents=body["contents"],
18
    config={
19
      "safety_settings": [body["safetySettings"]] # REST API uses camelCase
20
    }
21
  )
22
23
  # Parse the LLM response.
24
  parsed = b.parse.ExtractResume(res.text)
25
26
  # Fully parsed Resume type.
27
  print(parsed)

1
import asyncio
2
import json
3
import os
4
import httpx
5
from botocore.auth import SigV4Auth
6
from botocore.awsrequest import AWSRequest
7
import boto3
8
from baml_client import b
9
from urllib.parse import urlsplit
10
11
async def run():
12
  req = await b.request.ExtractResume("John Doe | Software Engineer | BSc in CS")
13
14
  body = req.body.json()
15
  # Optional: append your own messages before signing.
16
  body["messages"].append({
17
    "role": "system",
18
    "content": [{"text": "You must respond in JSON."}],
19
  })
20
  body_string = json.dumps(body)
21
  body_bytes = body_string.encode("utf-8")
22
23
  session = boto3.Session()
24
  credentials = session.get_credentials().get_frozen_credentials()
25
  region = (
26
    req.client_details.options.get("region")
27
    or os.environ.get("AWS_REGION")
28
    or os.environ.get("AWS_DEFAULT_REGION")
29
    or session.region_name
30
    or "us-east-1"
31
  )
32
33
  url = urlsplit(req.url)
34
35
  base_headers = {
36
    key: value
37
    for key, value in dict(req.headers).items()
38
    if value is not None
39
  }
40
41
  headers = {
42
    **base_headers,
43
    "content-type": "application/json",
44
    "accept": "application/json",
45
    "host": url.netloc,
46
  }
47
48
  aws_request = AWSRequest(
49
    method=req.method,
50
    url=req.url,
51
    data=body_bytes,
52
    headers=headers,
53
  )
54
  SigV4Auth(credentials, "bedrock", region).add_auth(aws_request)
55
56
  async with httpx.AsyncClient() as client:
57
    response = await client.post(
58
      req.url,
59
      headers={key: str(value) for key, value in aws_request.headers.items()},
60
      content=body_bytes,
61
    )
62
    if not response.is_success:
63
      raise RuntimeError(
64
        f"Bedrock request failed: {response.status_code} {response.text}"
65
      )
66
67
  payload = response.json()
68
  message = payload["output"]["message"]["content"][0]["text"]
69
  parsed = b.parse.ExtractResume(message)
70
  print(parsed)
71
72
asyncio.run(run())

1
import typing
2
from openai import AsyncOpenAI, AsyncStream
3
from openai.types.chat import ChatCompletionChunk
4
from baml_client import b
5
6
async def run():
7
  client = AsyncOpenAI()
8
9
  req = await b.stream_request.ExtractResume("John Doe | Software Engineer | BSc in CS")
10
11
  stream = typing.cast(
12
    AsyncStream[ChatCompletionChunk],
13
    await client.chat.completions.create(**req.body.json())
14
  )
15
16
  llm_response: list[str] = []
17
18
  async for chunk in stream:
19
    if len(chunk.choices) > 0 and chunk.choices[0].delta.content is not None:
20
      llm_response.append(chunk.choices[0].delta.content)
21
      # You can parse the partial responses as they come in.
22
      print(b.parse_stream.ExtractResume("".join(llm_response)))

1
import asyncio
2
import json
3
from openai import AsyncOpenAI
4
from baml_py import HTTPRequest as BamlHttpRequest
5
from baml_client import b
6
from baml_client import types
7
8
async def run():
9
  client = AsyncOpenAI()
10
11
  # Build the batch requests with BAML.
12
  john_req, jane_req = await asyncio.gather(
13
    b.request.ExtractResume("John Doe | Software Engineer | BSc in CS"),
14
    b.request.ExtractResume("Jane Smith | Data Scientist | PhD in Statistics"),
15
  )
16
17
  # Build the JSONL content.
18
  jsonl = to_openai_jsonl(john_req) + to_openai_jsonl(jane_req)
19
20
  # Create the batch input file.
21
  batch_input_file = await client.files.create(
22
    file=jsonl.encode("utf-8"),
23
    purpose="batch",
24
  )
25
26
  # Create the batch.
27
  batch = await client.batches.create(
28
    input_file_id=batch_input_file.id,
29
    endpoint="/v1/chat/completions",
30
    completion_window="24h",
31
    metadata={
32
      "description": "BAML Modular API Python Batch Example"
33
    },
34
  )
35
36
  # Wait for the batch to complete (exponential backoff).
37
  backoff = 2
38
  attempts = 0
39
  max_attempts = 5
40
41
  while True:
42
    batch = await client.batches.retrieve(batch.id)
43
    attempts += 1
44
45
    if batch.status == "completed":
46
        break
47
48
    if attempts >= max_attempts:
49
      try:
50
        await client.batches.cancel(batch.id)
51
      finally:
52
        raise Exception("Batch failed to complete in time")
53
54
    await asyncio.sleep(backoff)
55
    back_off *= 2
56
57
  # Retrieve the batch output file.
58
  output = await client.files.content(batch.output_file_id)
59
60
  # You can match the batch results using the BAML request IDs.
61
  expected = {
62
    john_req.id: types.Resume(
63
      name="John Doe",
64
      experience=["Software Engineer"],
65
      education=["BSc in CS"]
66
    ),
67
    jane_req.id: types.Resume(
68
      name="Jane Smith",
69
      experience=["Data Scientist"],
70
      education=["PhD in Statistics"]
71
    ),
72
  }
73
74
  resumes = {}
75
76
  for line in output.text.splitlines():
77
    result = json.loads(line)
78
    llm_response = result["response"]["body"]["choices"][0]["message"]["content"]
79
80
    parsed = b.parse.ExtractResume(llm_response)
81
    resumes[result["custom_id"]] = parsed
82
83
  print(resumes)
84
85
  # Should be equal.
86
  assert resumes == expected
87
88
89
def to_openai_jsonl(req: BamlHttpRequest) -> str:
90
  """ Helper that converts a BAML HTTP request to OpenAI JSONL format. """
91
  line = json.dumps({
92
    "custom_id": req.id, # Important for matching the batch results.
93
    "method": "POST",
94
    "url": "/v1/chat/completions",
95
    "body": req.body.json(),
96
  })
97
98
  return f"{line}\n"