Loading the notebook…
Loading the notebook…
Notebooks · Reliability and control · 60
Users paste images, audio, video, or PDFs anywhere in a prompt:
Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.
from pathlib import Path
import sys, json
ROOT = Path.cwd().resolve().parent if Path.cwd().name == 'notebooks' else Path.cwd().resolve()
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from shipit_agent.multimodal import (
MediaParser,
InMemoryMediaStore,
FileMediaStore,
StoredMedia,
MediaKind,
build_multimodal_message,
extract_media_refs,
)
# Local sample media — checked into notebooks/_media_samples/.
# We resolve to file:// URIs so the notebook runs offline.
SAMPLES = (Path.cwd() / '_media_samples') if (Path.cwd() / '_media_samples').exists() else (ROOT / 'notebooks' / '_media_samples')
def file_url(name):
return (SAMPLES / name).resolve().as_uri()
IMG_RED = file_url('sample_image.png')
IMG_BLUE = file_url('sample_image_blue.png')
AUDIO_WAV = file_url('sample_audio.wav')
VIDEO_MP4 = file_url('sample_video.mp4')
DOC_PDF = file_url('sample_doc.pdf')
print('Sample media resolved to:')
for label, url in [('image (red)', IMG_RED), ('image (blue)', IMG_BLUE),
('audio', AUDIO_WAV), ('video', VIDEO_MP4), ('document', DOC_PDF)]:
print(f' {label:14} {url}')MediaParser.parse() returns a ParsedPrompt with ordered Segments — either TextSegment or MediaSegment. The order matches the original prompt exactly.
parser = MediaParser(allowlist_domains=['*'])
prompt = (
'Hey, what is this [https://acme.com/cat.png] about? '
'Also compare it to .'
)
parsed = parser.parse(prompt)
for s in parsed.segments:
if s.is_text:
print(f'TEXT : {s.text!r}')
else:
print(f'MEDIA : {s.ref.kind.value} → {s.ref.url} (alt={s.ref.alt!r})')print(f'Total media refs : {len(parsed.media_refs)}')
print(f'Has media : {parsed.has_media}')
print(f'Cleaned text : {parsed.text_only!r}')The builder emits Anthropic-shape blocks. LiteLLM normalises this across Anthropic, OpenAI, Bedrock, and Gemini — you write it once.
msg = build_multimodal_message(parsed, role='user')
print(json.dumps(msg, indent=2))Notice how the content list interleaves: text, image, text, image, text. The vision LLM sees the cat image at exactly the position the user typed it.
In production you almost always want to lock media URLs to your own CDN. Both allowlist and denylist accept fnmatch globs (*.acme.com, evil.com, etc).
When a reference fails validation, the raw token is preserved as text — the user sees what was blocked rather than getting a silent drop.
strict_parser = MediaParser(
allowlist_domains=['*.acme.com', 'cdn.acme.com'],
denylist_domains=['evil.com'],
max_size_mb=10,
)
for url in [
'https://cdn.acme.com/ok.png',
'https://images.acme.com/ok.png',
'https://evil.com/bad.png',
'https://random.example.com/skipped.png',
]:
print(f'{url:50} → allowed={strict_parser.is_allowed_domain(url)}')# Disallowed refs are dropped from the segments list.
# The original token stays inline as plain text so the model still sees it.
mixed = strict_parser.parse(
'Compare [https://cdn.acme.com/ok.png] vs [https://evil.com/bad.png].'
)
for s in mixed.segments:
if s.is_text:
print(f'TEXT : {s.text!r}')
else:
print(f'MEDIA : {s.ref.url}')InMemoryMediaStoreWhen the user uploads a file (drag-drop, mobile capture, attachment), you store it and reference it by UUID. The agent resolves [media:<uuid>] automatically.
For production: implement the four-method MediaStore Protocol against your S3 bucket / DB / CDN.
store = InMemoryMediaStore()
asset_id = store.put(StoredMedia(
id='', # auto-generated UUID
url='https://internal.acme.com/uploads/screenshot.png',
mime='image/png',
alt='Bug screenshot from user',
))
print(f'Stored under id: {asset_id}')
store_parser = MediaParser(allowlist_domains=['*'], media_store=store)
parsed = store_parser.parse(
f'I am seeing a 500 error — here is the screenshot [media:{asset_id}].'
)
for s in parsed.segments:
if s.is_media:
print(f'Resolved → kind={s.ref.kind.value} url={s.ref.url} alt={s.ref.alt!r}')Real-life: PM drops two button mockups into a Slackbot and asks which is more accessible. The vision LLM sees both renders interleaved with the question and replies with a contrast/hit-target/focus-state critique.
design_prompt = (
'Compare these two button designs. Which is more accessible?\n\n'
'Variant A: \n'
'Variant B: '
)
design_msg = build_multimodal_message(
MediaParser(allowlist_domains=['figma.com']).parse(design_prompt)
)
print('content blocks:')
for b in design_msg['content']:
if b['type'] == 'text':
print(f" text : {b['text'][:60]!r}")
else:
print(f" {b['type']:7}: {b['source']['url']}")Drop-in code for production (commented — wire your own LLM):
from shipit_agent import Agent
from shipit_agent.llms.anthropic import AnthropicLLM
agent = Agent(
llm=AnthropicLLM(model='claude-sonnet-4-5'),
media_parser=MediaParser(allowlist_domains=['figma.com']),
)
result = agent.run(design_prompt)
print(result.output)
Real-life: A customer leaves a voicemail. The agent hears the actual audio (no Whisper roundtrip), tags urgency, and pulls out the concrete asks. Tone of voice and pacing inform the urgency tag.
voice_store = InMemoryMediaStore()
vm_id = voice_store.put(StoredMedia(
id='',
url='https://uploads.acme.com/voice/abc.m4a',
mime='audio/m4a',
alt='Customer voicemail (12 sec)',
))
voice_parser = MediaParser(allowlist_domains=['*'], media_store=voice_store)
voice_prompt = (
f'Triage this voicemail: [media:{vm_id}]. '
'Tag urgency (P0/P1/P2/P3) and pull out concrete asks. '
'Be brief.'
)
voice_msg = build_multimodal_message(voice_parser.parse(voice_prompt))
print(json.dumps(voice_msg, indent=2))Notice the audio content block — not image. The parser inferred the kind from the audio/m4a MIME we stored.
Provider compatibility for audio:
Audio + PDF in the same prompt. The model hears the kickoff, reads the roadmap, and reports drift between the two.
drift_parser = MediaParser(allowlist_domains=['*.acme.com'])
drift_msg = build_multimodal_message(drift_parser.parse(
'Compare what was said in this kickoff '
'[https://files.acme.com/kickoff.mp3] to the actual roadmap PDF '
'[https://files.acme.com/roadmap.pdf]. Where do the two diverge?'
))
for b in drift_msg['content']:
print(f"{b['type']:9}: {b.get('source', {}).get('url') or b['text'][:50]!r}")Real-life: Analyst pastes a 40-page earnings PDF and asks for the top three risks management mentioned. PDF goes in as a document content block; Anthropic / Bedrock / Gemini all read PDFs natively.
pdf_parser = MediaParser(allowlist_domains=['investor.acme.com'])
pdf_prompt = (
'Read [https://investor.acme.com/Q3-earnings.pdf] and summarise '
'the three biggest risks mentioned in management commentary. '
'Cite the page number for each.'
)
pdf_msg = build_multimodal_message(pdf_parser.parse(pdf_prompt))
for b in pdf_msg['content']:
print(f"{b['type']:9}: {b.get('source', {}).get('url') or b['text'][:80]!r}")Real-life: User shares a chart screenshot, asks the agent to read it, then verify the numbers via a separate web_search tool. Multimodal blocks compose with regular tools — same agent loop, both visible in the trace.
chained_parser = MediaParser(allowlist_domains=['*'])
chained = chained_parser.parse(
'What is in this image [https://acme.com/Q3-revenue.png]? '
'Then look up the underlying SEC filing via web_search and verify the numbers.'
)
chained_msg = build_multimodal_message(chained)
print(json.dumps(chained_msg, indent=2))Drop-in code for production:
from shipit_agent import Agent
from shipit_agent.tools import WebSearchTool
agent = Agent(
llm=vision_llm,
tools=[WebSearchTool()],
media_parser=chained_parser,
)
result = agent.run(
'What is in this image [https://acme.com/Q3-revenue.png]? '
'Then look up the SEC filing via web_search and verify the numbers.'
)
# Trace: vision → read chart → web_search('Acme Q3 revenue 10-Q') → verify → reply
The kitchen sink — image, audio, video, document all routed to the right block type.
kitchen_sink = (
'Tag this asset folder. '
'Image: [https://acme.com/cover.png]. '
'Audio: [https://acme.com/voice.mp3]. '
'Video: [https://acme.com/clip.mp4]. '
'Doc: [https://acme.com/spec.pdf].'
)
msg = parser.parse(kitchen_sink)
blocks = build_multimodal_message(msg)
for block in blocks['content']:
if block['type'] == 'text':
print(f"TEXT : {block['text'][:60]!r}")
else:
print(f"{block['type'].upper():9}: {block['source']['url']}")Every example above used remote URLs. The repo also ships a tiny
_media_samples/ directory so you can verify parsing/builder against
real local media without network access. Image, audio, video, and
PDF — one of each.
offline_prompt = (
f'Asset audit:\n'
f'- Cover image: [{IMG_RED}]\n'
f'- Voice memo: [{AUDIO_WAV}]\n'
f'- Demo clip: [{VIDEO_MP4}]\n'
f'- Spec doc: [{DOC_PDF}]\n'
f'Tag each asset with its detected kind.'
)
offline_parser = MediaParser(allowlist_domains=['*'])
offline_msg = build_multimodal_message(offline_parser.parse(offline_prompt))
print('Detected blocks:')
for b in offline_msg['content']:
if b['type'] == 'text':
print(f" text · {b['text'].strip()[:60]!r}")
else:
print(f" {b['type']:9}· {b['source']['url'].split('/')[-1]}")
FileMediaStoreAcross restarts you usually want a real backing store. Use FileMediaStore for local dev, swap in your own S3/DB-backed implementation for production.
import tempfile, os
tmp = tempfile.NamedTemporaryFile(suffix='.json', delete=False)
tmp.close()
fstore = FileMediaStore(tmp.name)
uid = fstore.put(StoredMedia(
id='', url='https://acme.com/avatar.png', mime='image/png', alt='User avatar',
))
# Simulate restart — new process, same file
fstore2 = FileMediaStore(tmp.name)
loaded = fstore2.resolve(uid)
print(f'Reloaded after restart: {loaded.url} (alt={loaded.alt!r})')
os.unlink(tmp.name)extract_media_refsIf you only need the list of references and don't care about segment ordering, this one-shot is handy for logging and audit trails.
refs = extract_media_refs(
'See [https://acme.com/a.png] and .',
allowlist_domains=['*.acme.com'],
)
for r in refs:
print(f'{r.kind.value:8} {r.url} alt={r.alt!r}')Multimodal composes cleanly with the rest of the runtime:
{subject, colors, confidence})MediaParser — parses the prompt, returns a ParsedPromptMediaReference / MediaSegment / TextSegment — typed modelMediaKind — IMAGE / AUDIO / VIDEO / DOCUMENT / UNKNOWNMediaStore (Protocol), InMemoryMediaStore, FileMediaStore — plug your own backend or use the bundled onesbuild_multimodal_message — ordered Anthropic-shape content blocksextract_media_refs — quick one-shot if you don't need segmentsAll parts are independently testable. 82 tests in tests/test_multimodal.py.