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!

Data Handling

Data Handling

The Data Class

Accessing Data Content

Working with Responses

Media Type Helpers

Streaming Data

Need Help?

On this page