Data Handling
Working with various data types in the Agentuity Python SDK
Data Handling
The Agentuity Python SDK provides a flexible system for working with different types of data, from text and JSON to binary formats like images and audio.
The Data Class
The Data class serves as a container for all types of content:
from agentuity.server.data import Data, StringStreamReader, BytesStreamReader
# Create a text Data object
text_data = Data("text/plain", StringStreamReader("Hello, world!"))
# Create a binary Data object
binary_data = Data("application/octet-stream", BytesStreamReader(b"\x00\x01\x02\x03"))Accessing Data Content
Data content can be accessed in various formats:
async def process_data(data: Data):
# Get as text
text = await data.text()
# Get as JSON
json_data = await data.json()
# Get as binary
binary = await data.binary()
# Get as base64 encoded string
base64_str = await data.base64()
# Access as a stream
stream = await data.stream()
async for chunk in stream:
process_chunk(chunk)Working with Responses
The AgentResponse class provides methods for various content types:
async def run(request, response, context):
# Text response
return response.text("Hello, world!")
# JSON response
return response.json({"message": "Hello, world!"})
# Markdown response
return response.markdown("# Hello, world!")
# Binary response
return response.binary(b"\x00\x01\x02\x03", "application/octet-stream")Media Type Helpers
The SDK includes helpers for common media types:
async def run(request, response, context):
# Image responses
with open("image.png", "rb") as f:
image_data = f.read()
return response.png(image_data)
# Other image formats
return response.jpeg(jpeg_data)
return response.gif(gif_data)
return response.webp(webp_data)
# Document formats
return response.pdf(pdf_data)
# Audio/Video formats
return response.mp3(mp3_data)
return response.mp4(mp4_data)
return response.wav(wav_data)
return response.ogg(ogg_data)
return response.webm(webm_data)
return response.m4a(m4a_data)Streaming Data
For large responses or real-time data, use streaming:
async def run(request, response, context):
# Stream data from a generator
async def generator():
for i in range(10):
yield f"Item {i}\n"
# Return the stream
return response.stream(generator(), contentType="text/plain")
# Transform items during streaming
def transform(item):
return f"Transformed: {item}"
return response.stream(generator(), transform=transform)
# Stream responses from another agent
other_response = await context.get_agent("other_agent").run(request.data)
return response.stream(other_response)Need Help?
Join our Community for assistance or just to hang with other humans building agents.
Send us an email at hi@agentuity.com if you'd like to get in touch.
Please Follow us on
If you haven't already, please Signup for your free account now and start building your first agent!