SDKs/Python SDK

API Reference

Complete reference documentation for the Agentuity Python SDK, including request handling, response types, data storage, and agent communication.

This section provides detailed documentation for the Agentuity Python SDK API, including method signatures, parameters, return values, and example usage.

Table of Contents

Agent Lifecycle
Storage APIs
Agent Communication
Response Types
Request Handling
Logging
Telemetry

Note: This documentation applies to Agentuity Python SDK version 1.0 and above.

Welcome Function

The Agentuity SDK allows you to customize the initial appearance of DevMode when it starts interacting with your agents by defining a welcome() function in your agent module. This function should return a dictionary containing a welcome message and optional example prompts.

welcome()

Defines a welcome message and optional example prompts for DevMode.

Return Value

Returns a dictionary with a welcome message and optional prompts.

def welcome():
    return {
        "welcome": "Welcome message to display",
        "prompts": [
            {
                "data": "Example prompt text",
                "contentType": "text/plain"
            }
        ]
    }

Example

def welcome():
    return {
        "welcome": "Welcome to my Python Agent! How can I help you today?",
        "prompts": [
            {
                "data": "What can you do?",
                "contentType": "text/plain"
            }
        ]
    }

Email Processing

The Agentuity SDK provides comprehensive email processing capabilities, including parsing incoming emails, handling attachments, and sending replies.

Email Class

The Email class represents an incoming email with methods to access its properties and send replies.

`sendReply(request, context, subject=None, text=None, html=None, attachments=None)`

Send a reply to the incoming email using the Agentuity email API.

Parameters

request (AgentRequest): The triggering agent request, used to extract metadata such as email-auth-token
context (AgentContext): The agent context, used to get the base_url and agentId
subject (str, optional): Subject of the reply. Defaults to 'Re: <original subject>'
text (str, optional): Plain text body of the reply
html (str, optional): HTML body of the reply
attachments (List[OutgoingEmailAttachmentInterface], optional): List of attachments to include

Return Value

Returns None on successful send.

Example

from agentuity import AgentRequest, AgentResponse, AgentContext
from agentuity.io.email import EmailAttachment
 
async def handler(request: AgentRequest, response: AgentResponse, context: AgentContext):
    # Get the incoming email
    email = request.data.email()
    
    # Create an attachment
    attachment = EmailAttachment(
        filename="response.txt",
        data="Thank you for your inquiry!",
        content_type="text/plain"
    )
    
    # Send a reply
    await email.sendReply(
        request=request,
        context=context,
        subject="Re: Your inquiry",
        text="Thank you for contacting us. We'll get back to you soon.",
        html="<p>Thank you for contacting us. We'll get back to you soon.</p>",
        attachments=[attachment]
    )
    
    return response.json({"status": "Reply sent successfully"})

EmailAttachment Class

The EmailAttachment class represents an outgoing email attachment with streaming data support.

Constructor

EmailAttachment(filename: str, data: DataLike, content_type: str = None)

Parameters

filename (str): The filename of the attachment
data (DataLike): The attachment data (can be string, bytes, or Data object)
content_type (str, optional): MIME type of the attachment

Properties

filename (str): The filename of the attachment
data(): Returns the Data object containing the attachment content

IncomingEmailAttachment Class

The IncomingEmailAttachment class represents an attachment from an incoming email with support for large attachments via streaming.

Properties

filename (str): The filename of the attachment
content_disposition (str): The content disposition ('attachment' or 'inline')

Methods

`async data()`

Asynchronously retrieves the attachment data as a Data object with streaming support for large files.

Return Value

Returns a Data object that streams the attachment content.

Example

async def handler(request: AgentRequest, response: AgentResponse, context: AgentContext):
    email = request.data.email()
    
    # Process attachments
    for attachment in email.attachments:
        context.logger.info(f"Processing attachment: {attachment.filename}")
        
        # Stream large attachment data
        data = await attachment.data()
        content = await data.binary()
        
        # Process the attachment content
        context.logger.info(f"Attachment size: {len(content)} bytes")
    
    return response.json({"attachments_processed": len(email.attachments)})

Agent Lifecycle

The Agentuity SDK provides a structured way to define and handle agents. An agent consists of a handler function, with its configuration managed by the Agentuity CLI.

Agent Handler

The agent handler is a function that processes requests and returns responses:

from agentuity import AgentRequest, AgentResponse, AgentContext
 
async def agent_handler(
    request: AgentRequest,
    response: AgentResponse,
    context: AgentContext
) -> AgentResponseType:
    """
    Handler function for an agent.
    
    Args:
        request: An AgentRequest object containing the request data
        response: An AgentResponse object for creating responses
        context: An AgentContext object providing access to various capabilities
        
    Returns:
        An AgentResponseType object representing the response
    """
    return response.json({"message": "Hello, world!"})

Parameters

request: An AgentRequest object containing the request data
response: An AgentResponse object for creating responses
context: An AgentContext object providing access to various capabilities

Return Value

The handler function should return an AgentResponseType object.

Example

from agentuity import AgentRequest, AgentResponse, AgentContext
 
# Agent handler function
async def handler(request: AgentRequest, response: AgentResponse, context: AgentContext):
    try:
        # Get the request data
        data = request.data.json
        name = data.get("name")
        
        # Log the request
        context.logger.info(f"Received greeting request for {name}")
        
        # Return a personalized greeting
        return response.json({
            "message": f"Hello, {name}! Welcome to Agentuity."
        })
    except Exception as error:
        # Handle errors
        context.logger.error("Error processing request", error)
        return response.json({"error": "Failed to process request"})

Storage APIs

The Agentuity SDK provides three storage APIs: Key-Value Storage, Vector Storage, and Object Storage.

Key-Value Storage

The Key-Value Storage API provides a simple way to store and retrieve data. It is accessed through the context.kv object.

`async get(name: str, key: str) -> "DataResult"`

Retrieves a value from the key-value storage.

Parameters

name: The name of the key-value storage
key: The key to retrieve the value for

Return Value

Returns a DataResult object that has an exists property to check if the value exists and a data property with accessors like data.json and data.text.

Example

from agentuity.sdk import AgentContext
 
# Retrieve a value from key-value storage with error handling
try:
    value = await context.kv.get("user-preferences", "user-123")
    if value.exists:
        # Access data using the appropriate accessor
        user_prefs = value.data.json
        context.logger.info(f"User preferences: {user_prefs}")
        
        # Or access as text if needed
        # text_data = value.data.text
        # context.logger.info(f"User preferences (text): {text_data}")
    else:
        context.logger.info("User preferences not found, using defaults")
        user_prefs = {"theme": "light"}
except Exception as e:
    context.logger.error(f"Failed to retrieve user preferences: {str(e)}")
    # Use default value or re-raise exception
    user_prefs = {"theme": "light"}

`async set(name: str, key: str, value: Union[str, int, float, bool, list, dict, bytes, "Data"], params: Optional[dict] = None) -> None`

Stores a value in the key-value storage.

Parameters

name: The name of the key-value storage
key: The key to store the value under
value: The value to store (Union[str, int, float, bool, list, dict, bytes, "Data"])
params (optional): Can have ttl (time-to-live in seconds)

Return Value

None

Example

import json
 
# Store a string value
await context.kv.set("user-preferences", "user-123", json.dumps({"theme": "dark"}))
 
# Store a dictionary value (automatically serialized to JSON)
await context.kv.set("user-preferences", "user-123", {"theme": "dark"})
 
# Store binary data
binary_data = bytes([1, 2, 3, 4])
await context.kv.set("user-data", "user-123", binary_data)
 
# Store a value with TTL (expires after 1 hour)
await context.kv.set("session", "user-123", "active", {"ttl":3600})

`async delete(name: str, key: str) -> None`

Deletes a value from the key-value storage.

Parameters

name: The name of the key-value storage
key: The key to delete

Return Value

None

Example

# Delete a value
await context.kv.delete("user-preferences", "user-123")

Vector Storage

The Vector Storage API provides a way to store and search for data using vector embeddings. It is accessed through the context.vector object.

`async upsert(name: str, *documents: VectorUpsertParams) -> list[str]`

Inserts or updates vectors in the vector storage.

Parameters

name: The name of the vector storage
documents: One or more documents to upsert, each with either embeddings or text

Return Value

Returns a list of string IDs for the upserted vectors.

Example

from agentuity.sdk import AgentContext
from typing import List, Dict, Any
 
async def index_products(context: AgentContext, products: List[Dict[str, Any]]) -> List[str]:
    """
    Index product descriptions in vector storage for semantic search.
    
    Args:
        context: The agent context
        products: List of product dictionaries with name, description, and category
        
    Returns:
        List of vector IDs for the indexed products
    """
    # Prepare documents for vector storage
    documents = []
    for product in products:
        documents.append({
            "document": product["description"],
            "metadata": {
                "product_id": product["id"],
                "name": product["name"],
                "category": product["category"],
                "price": product["price"]
            }
        })
    
    # Upsert documents to vector storage
    try:
        ids = await context.vector.upsert("product-descriptions", *documents)
        context.logger.info(f"Indexed {len(ids)} products in vector storage")
        return ids
    except Exception as e:
        context.logger.error(f"Failed to index products: {str(e)}")
        raise
 
# Simple example
# Upsert documents with text
ids = await context.vector.upsert(
    "product-descriptions",
    {"document": "Ergonomic office chair with lumbar support", "metadata": {"category": "furniture"}},
    {"document": "Wireless noise-cancelling headphones", "metadata": {"category": "electronics"}}
)
 
# Upsert documents with embeddings
ids2 = await context.vector.upsert(
    "product-embeddings",
    {"embeddings": [0.1, 0.2, 0.3, 0.4], "metadata": {"productId": "123"}},
    {"embeddings": [0.5, 0.6, 0.7, 0.8], "metadata": {"productId": "456"}}
)

`async search(name: str, params: VectorSearchParams) -> list[VectorSearchResult]`

Searches for vectors in the vector storage.

Parameters

name: The name of the vector storage
params: Search parameters including query, limit, similarity threshold, and metadata filters

Return Value

Returns a list of search results, each containing an ID, metadata, and distance score.

Example

from agentuity.sdk import AgentContext
from typing import List, Dict, Any
 
# Search for similar products with error handling
try:
    results = await context.vector.search("product-descriptions", {
        "query": "comfortable office chair",
        "limit": 5,
        "similarity": 0.7,
        "metadata": {"category": "furniture"}
    })
    
    # Process search results
    if results:
        context.logger.info(f"Found {len(results)} matching products")
        for result in results:
            print(f"Product ID: {result.id}, Similarity: {result.distance}")
            print(f"Metadata: {result.metadata}")
    else:
        context.logger.info("No matching products found")
except Exception as e:
    context.logger.error(f"Vector search failed: {str(e)}")
    # Handle the error appropriately

`async delete(name: str, key: str) -> int`

Deletes a vector from the vector storage.

Parameters

name: The name of the vector storage
key: The ID of the vector to delete

Return Value

Returns the number of vectors that were deleted (0 or 1).

Example

from agentuity.sdk import AgentContext
 
# Delete a single vector with error handling
try:
    deleted_count = await context.vector.delete("product-descriptions", "id1")
    context.logger.info(f"Deleted {deleted_count} vector(s)")
except Exception as e:
    context.logger.error(f"Failed to delete vector: {str(e)}")
    # Handle the error appropriately

Object Storage

The Object Storage API provides a way to store and retrieve objects (files, documents, media) with support for public URL generation. It is accessed through the context.objectstore object.

`async get(bucket: str, key: str) -> DataResult`

Retrieves an object from the object storage.

Parameters

bucket: The bucket to get the object from
key: The key of the object to get

Return Value

Returns a DataResult object that has an exists property to check if the object exists and a data property containing the object data if found.

Example

from agentuity import AgentContext
 
# Retrieve an object from object storage with error handling
try:
    result = await context.objectstore.get("user-uploads", "profile-123.jpg")
    if result.exists:
        # Access data using the Data interface methods
        image_data = await result.data.binary()
        context.logger.info(f"Image size: {len(image_data)} bytes")
    else:
        context.logger.info("Image not found")
except Exception as e:
    context.logger.error(f"Failed to retrieve object: {str(e)}")
    # Handle the error appropriately

`async put(bucket: str, key: str, data: DataLike, params: Optional[ObjectStorePutParams] = None) -> None`

Stores an object in the object storage.

Parameters

bucket: The bucket to put the object into
key: The key of the object to put
data: The data to store (can be string, bytes, or other DataLike types)
params (optional): Additional parameters for the object. The following properties map to canonical HTTP headers: content_type (Content-Type), content_encoding (Content-Encoding), cache_control (Cache-Control), content_disposition (Content-Disposition), and content_language (Content-Language). The optional metadata property accepts a dictionary of arbitrary key-value pairs that will be stored as metadata with the object but not returned as part of HTTP results when the object is fetched. The content_type parameter is optional - if not provided, the content type will be automatically detected based on the data type (e.g., objects will be set to application/json). If the content type cannot be determined, it defaults to application/octet-stream.

Return Value

Returns None when the object has been stored successfully.

Example

from agentuity import AgentContext
from agentuity.server.objectstore import ObjectStorePutParams
 
# Store a text file with explicit content type
await context.objectstore.put(
    "documents", 
    "readme.txt", 
    "Hello, world!",
    ObjectStorePutParams(content_type="text/plain")
)
 
# Store an object (content type auto-detected as application/json)
user_data = {"name": "John", "age": 30}
await context.objectstore.put("users", "user-123.json", user_data)
 
# Store binary data
image_data = bytes([/* image bytes */])
await context.objectstore.put(
    "images", 
    "photo.jpg", 
    image_data,
    ObjectStorePutParams(content_type="image/jpeg")
)
 
# Store without specifying content type (will default to application/octet-stream if unknown)
await context.objectstore.put("files", "unknown-data", some_data)
 
# Store with custom encoding and cache control
await context.objectstore.put(
    "compressed", 
    "data.gz", 
    compressed_data,
    ObjectStorePutParams(
        content_type="application/octet-stream",
        content_encoding="gzip",
        cache_control="max-age=3600"
    )
)
 
# Store with metadata (not returned in HTTP results when fetched)
await context.objectstore.put(
    "uploads", 
    "document.pdf", 
    pdf_data,
    ObjectStorePutParams(
        content_type="application/pdf",
        content_disposition="attachment; filename=\"report.pdf\"",
        content_language="en-US",
        metadata={
            "user-id": "12345",
            "upload-source": "web-app",
            "version": "2.0"
        }
    )
)

`async delete(bucket: str, key: str) -> bool`

Deletes an object from the object storage.

Parameters

bucket: The bucket to delete the object from
key: The key of the object to delete

Return Value

Returns a boolean indicating whether the object was deleted (True) or didn't exist (False).

Example

from agentuity import AgentContext
 
# Delete an object with error handling
try:
    was_deleted = await context.objectstore.delete("user-uploads", "old-file.pdf")
    if was_deleted:
        context.logger.info("File deleted successfully")
    else:
        context.logger.info("File did not exist")
except Exception as e:
    context.logger.error(f"Failed to delete object: {str(e)}")
    # Handle the error appropriately

`async create_public_url(bucket: str, key: str, expires_duration: Optional[int] = None) -> str`

Creates a time-limited public URL for accessing an object.

Parameters

bucket: The bucket containing the object
key: The key of the object
expires_duration (optional): Duration in milliseconds until the URL expires. Defaults to 1 hour (3600000ms) if not provided. Minimum value is 1 minute (60000ms) - values less than 1 minute will be set to 1 minute.

Return Value

Returns a public URL string that can be used to access the object.

Example

from agentuity import AgentContext
 
# Create a public URL that expires in 1 hour
try:
    public_url = await context.objectstore.create_public_url(
        "user-uploads", 
        "document.pdf", 
        expires_duration=3600000  # 1 hour in milliseconds
    )
    
    # Share the URL with users
    context.logger.info(f"Download link: {public_url}")
    
    # Create a URL with default expiration (1 hour)
    default_url = await context.objectstore.create_public_url("images", "photo.jpg")
    
    # Create a URL with minimum expiration (1 minute)
    short_url = await context.objectstore.create_public_url(
        "temp-files", 
        "quick-access.txt", 
        expires_duration=60000  # 1 minute in milliseconds
    )
    
    # Values less than 1 minute will be automatically set to 1 minute
    auto_min_url = await context.objectstore.create_public_url(
        "temp-files", 
        "auto-min.txt", 
        expires_duration=30000  # Will be set to 60000ms (1 minute)
    )
    
except Exception as e:
    context.logger.error(f"Failed to create public URL: {str(e)}")
    # Handle the error appropriately

Agent Communication

The Agentuity SDK allows agents to communicate with each other through the context.get_agent() method and agent redirection.

`async get_agent(params: GetAgentRequestParams) -> RemoteAgent`

Retrieves a handle to a remote agent that can be invoked.

Parameters

params: Parameters to identify the agent, either by ID or by name and project ID

Return Value

Returns a RemoteAgent object that can be used to invoke the agent.

Example

from agentuity.sdk import AgentContext, RemoteAgent
 
# Get an agent by ID
agent = await context.get_agent({"id": "agent-123"})
 
# Get an agent by name
agent2 = await context.get_agent({
    "name": "data-processing-agent",
    "project_id": "project-456"
})
 
# Invoke the agent with error handling
try:
    result = await agent.run({"data": "process this"}, "application/json")
    # Process the result
    print(f"Agent response: {result}")
except Exception as e:
    context.logger.error(f"Failed to invoke agent: {str(e)}")
    # Handle the error appropriately

Agent Handoff

The response.handoff() method allows an agent to handoff the request to another agent.

`handoff(agent: GetAgentRequestParams, payload: dict | bytes | str = None, content_type: str = None, metadata: dict = None) -> AgentRedirectResponse`

Redirects the current request to another agent.

Handoff

Handoff is currently only supported for handoff to other agents in the same project. However, we are working on remote agent handoff and should have that working soon.

Parameters

agent: Parameters to identify the target agent
payload (optional): The payload to send to the target agent
content_type (optional): The content type of the payload
metadata (optional): Additional metadata to include with the request

Return Value

Returns an AgentRedirectResponse object.

Example

# Handoff to another agent
return response.handoff(
    {"name": "data-processing-agent"},
    {"data": "process this"},
    "application/json",
    {"source": "web-agent"}
)

Response Types

The Agentuity SDK provides various methods for creating different types of responses through the response object.

JSON Responses

`json(data: dict, metadata: dict = None) -> AgentResponseType`

Creates a JSON response.

Parameters

data: The JSON data to include in the response
metadata (optional): Additional metadata to include with the response

Return Value

Returns an AgentResponseType object with the JSON data.

Example

return response.json({
    "message": "Success",
    "data": {"id": 123, "name": "Example"}
})

Text Responses

`text(data: str, metadata: dict = None) -> AgentResponseType`

Creates a text response.

Parameters

data: The text to include in the response
metadata (optional): Additional metadata to include with the response

Return Value

Returns an AgentResponseType object with the text data.

Example

return response.text("Hello, world!")

Binary Responses

`binary(data: bytes, metadata: dict = None) -> AgentResponseType`

Creates a binary response.

Parameters

data: The binary data to include in the response
metadata (optional): Additional metadata to include with the response

Return Value

Returns an AgentResponseType object with the binary data.

Example

binary_data = bytes([1, 2, 3, 4])
return response.binary(binary_data, {"filename": "data.bin"})

Media Type Responses

The SDK provides specialized methods for various media types:

pdf(data: bytes, metadata: dict = None) -> AgentResponseType
png(data: bytes, metadata: dict = None) -> AgentResponseType
jpeg(data: bytes, metadata: dict = None) -> AgentResponseType
gif(data: bytes, metadata: dict = None) -> AgentResponseType
webp(data: bytes, metadata: dict = None) -> AgentResponseType
mp3(data: bytes, metadata: dict = None) -> AgentResponseType
mp4(data: bytes, metadata: dict = None) -> AgentResponseType
wav(data: bytes, metadata: dict = None) -> AgentResponseType
ogg(data: bytes, metadata: dict = None) -> AgentResponseType
data(data: Any, content_type: str, metadata: dict = None) -> AgentResponseType
markdown(content: str, metadata: dict = None) -> AgentResponseType

Each method works similarly to the binary() method but sets the appropriate content type. The data method allows setting specific data with an exact content type, while the markdown method provides a convenient way to return markdown content.

Example

# Return a PNG image
return response.png(image_data, {"filename": "chart.png"})
 
# Return an MP3 audio file
return response.mp3(audio_data, {"duration": 120})

HTML Responses

`html(data: str, metadata: dict = None) -> AgentResponseType`

Creates an HTML response.

Parameters

data: The HTML content to include in the response
metadata (optional): Additional metadata to include with the response

Return Value

Returns an AgentResponseType object with the HTML content.

Example

return response.html("<h1>Hello, world!</h1><p>This is an HTML response.</p>")

Empty Responses

`empty(metadata: dict = None) -> AgentResponseType`

Creates an empty response.

Parameters

metadata (optional): Additional metadata to include with the response

Return Value

Returns an AgentResponseType object with no payload.

Example

return response.empty()

Request Handling

The Agentuity SDK provides various methods for accessing request data through the request object.

Accessing Request Data

`trigger() -> str`

Gets the trigger type of the request.

Return Value

Returns a string representing the trigger type (webhook, cron, manual, agent, etc.).

Example

trigger_type = request.trigger()
print(f"Request triggered by: {trigger_type}")

`metadata(key: str, default_value = None) -> Any`

Gets metadata associated with the request.

Parameters

key: The metadata key to retrieve
default_value (optional): A default value to return if the key does not exist

Return Value

Returns the metadata value for the specified key, or the default value if the key does not exist.

Example

user_id = request.metadata("userId", "anonymous")
print(f"User ID: {user_id}")

`json() -> dict`

Gets the payload of the request as a JSON object.

Return Value

Returns the request payload as a dictionary.

Example

data = request.data.json
print(f"Request data: {data}")

`text() -> str`

Gets the payload of the request as a string.

Return Value

Returns the request payload as a string.

Example

text = request.data.text
print(f"Request text: {text}")

`binary() -> bytes`

Gets the payload of the request as bytes.

Return Value

Returns the request payload as bytes.

Example

binary_data = request.binary()
print(f"Binary data size: {len(binary_data)} bytes")

Media-Specific Methods

The SDK provides specialized methods for various media types:

pdf() -> bytes
png() -> bytes
jpeg() -> bytes
gif() -> bytes
webp() -> bytes
mp3() -> bytes
wav() -> bytes
ogg() -> bytes
email() -> Email

Each method returns the request payload as bytes with the appropriate content type validation, except for email() which returns an Email object.

Example

# Get an image from the request
image = request.png()
# Process the image...
 
# Get audio from the request
audio = request.mp3()
# Process the audio...

`email() -> Email`

Gets the payload of the request as an Email object. This method validates that the content type is message/rfc822 before parsing.

Return Value

Returns an Email object with parsed email data and accessor methods.

Example

# Get email from the request (content type must be message/rfc822)
email = await request.data.email()
 
# Access email properties
print(f"Subject: {email.subject}")
print(f"From: {email.from_name} <{email.from_email}>")
print(f"To: {email.to}")
print(f"Date: {email.date}")
print(f"Message ID: {email.messageId}")
 
# Access email content
print(f"Text content: {email.text}")
print(f"HTML content: {email.html}")
 
# Access attachments
for attachment in email.attachments:
    print(f"Attachment: {attachment}")
 
# Convert to dictionary
email_dict = email.to_dict()

Email Class

The Email class provides comprehensive email parsing capabilities and is returned by the email() method on request data.

Email Properties

All email properties are accessible as attributes on the Email object:

`subject -> str`

Returns the subject line of the email.

`from_email -> str | None`

Returns the sender's email address. Handles both simple email strings and tuple formats like ('Name', 'email@domain.com').

`from_name -> str | None`

Returns the sender's display name if available.

`to -> str | None`

Returns the recipient's email address. Handles both simple email strings and tuple formats.

`date -> datetime | None`

Returns the email date as a datetime object.

`messageId -> str`

Returns the unique message ID of the email.

`headers -> dict`

Returns all email headers as a dictionary.

`text -> str`

Returns the plain text content of the email.

`html -> str`

Returns the HTML content of the email.

`attachments -> list`

Returns a list of email attachments.

Email Methods

`to_dict() -> dict`

Converts the Email object to a dictionary containing all email data.

Return Value

Returns a dictionary with keys: subject, from_email, from_name, to, date, messageId, headers, text, html, attachments.

Example

email_data = email.to_dict()
print(email_data)
# {
#   "subject": "Welcome to Agentuity",
#   "from_email": "support@agentuity.com",
#   "from_name": "Agentuity Support",
#   "to": "user@example.com",
#   "date": "2025-05-24T10:30:00Z",
#   "messageId": "<abc123@agentuity.com>",
#   "headers": {...},
#   "text": "Welcome to our platform...",
#   "html": "<html>...</html>",
#   "attachments": []
# }

Dictionary-Style Access

The Email object also supports dictionary-style access to its properties:

# Access via dictionary notation
subject = email["subject"]
from_email = email["from_email"]
 
# Iterate over email data
for key, value in email.items():
    print(f"{key}: {value}")
 
# Get available keys
keys = email.keys()

Logging

The Agentuity SDK provides logging functionality through the context.logger object.

Logger Interface

The Logger class defines the following methods:

class Logger:
    """
    Logger for agent execution.
    """
    def debug(self, message: str, *args, **kwargs) -> None:
        """Log a debug message."""
        pass
        
    def info(self, message: str, *args, **kwargs) -> None:
        """Log an informational message."""
        pass
        
    def warn(self, message: str, *args, **kwargs) -> None:
        """Log a warning message."""
        pass
        
    def error(self, message: str, *args, **kwargs) -> None:
        """Log an error message."""
        pass
        
    def child(self, **kwargs) -> 'Logger':
        """Create a child logger with additional context."""
        pass

Logging Methods

`debug(message: str, *args, **kwargs) -> None`

Logs a debug message.

Parameters

message: The message to log
args, kwargs: Additional arguments to include in the log

Example

context.logger.debug("Processing request", request_id="123")

`info(message: str, *args, **kwargs) -> None`

Logs an informational message.

Parameters

message: The message to log
args, kwargs: Additional arguments to include in the log

Example

context.logger.info("Request processed successfully", request_id="123")

`warn(message: str, *args, **kwargs) -> None`

Logs a warning message.

Parameters

message: The message to log
args, kwargs: Additional arguments to include in the log

Example

context.logger.warn("Resource not found", resource_id="456")

`error(message: str, *args, **kwargs) -> None`

Logs an error message.

Parameters

message: The message to log
args, kwargs: Additional arguments to include in the log

Example

try:
    # Some code that might raise an exception
    result = process_data()
except Exception as e:
    context.logger.error("Failed to process request", error=str(e))

Creating Child Loggers

`child(**kwargs) -> Logger`

Creates a child logger with additional context.

Parameters

kwargs: Additional context to include in all logs from the child logger

Return Value

Returns a new Logger instance with the additional context.

Example

request_logger = context.logger.child(request_id="123", user_id="456")
request_logger.info("Processing request")  # Includes request_id and user_id

Telemetry

The Agentuity SDK integrates with OpenTelemetry for tracing and metrics.

Tracing

The SDK provides access to OpenTelemetry tracing through the context.tracer object.

Example

from agentuity.sdk import AgentContext
from opentelemetry.trace.status import Status, StatusCode
 
# Create a span
async with context.tracer.start_as_current_span("process-data") as span:
    try:
        # Add attributes to the span
        span.set_attribute("userId", "123")
        
        # Perform some work
        result = await process_data()
        
        # Add events to the span
        span.add_event("data-processed", {"itemCount": len(result)})
        
        return result
    except Exception as error:
        # Record the error
        span.record_exception(error)
        span.set_status(Status(StatusCode.ERROR))
        context.logger.error(f"Error processing data: {str(error)}")
        raise error

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!

Python SDK

Documentation for the Agentuity Python SDK

Core Concepts

Learn about the fundamental concepts of the Agentuity Python SDK

On this page

Table of Contents Welcome Function welcome()Return Value Example Email Processing Email Class sendReply(request, context, subject=None, text=None, html=None, attachments=None)Parameters Return Value Example EmailAttachment Class Constructor Parameters Properties IncomingEmailAttachment Class Properties Methods async data()Return Value Example Agent Lifecycle Agent Handler Parameters Return Value Example Storage APIs Key-Value Storage async get(name: str, key: str) -> "DataResult"Parameters Return Value Example

async set(name: str, key: str, value: Union[str, int, float, bool, list, dict, bytes, "Data"], params: Optional[dict] = None) -> None

Parameters Return Value Example async delete(name: str, key: str) -> NoneParameters Return Value Example Vector Storage async upsert(name: str, *documents: VectorUpsertParams) -> list[str]Parameters Return Value Example async search(name: str, params: VectorSearchParams) -> list[VectorSearchResult]Parameters Return Value Example async delete(name: str, key: str) -> intParameters Return Value Example Object Storage async get(bucket: str, key: str) -> DataResultParameters Return Value Example async put(bucket: str, key: str, data: DataLike, params: Optional[ObjectStorePutParams] = None) -> NoneParameters Return Value Example async delete(bucket: str, key: str) -> boolParameters Return Value Example async create_public_url(bucket: str, key: str, expires_duration: Optional[int] = None) -> strParameters Return Value Example Agent Communication async get_agent(params: GetAgentRequestParams) -> RemoteAgentParameters Return Value Example Agent Handoff

handoff(agent: GetAgentRequestParams, payload: dict | bytes | str = None, content_type: str = None, metadata: dict = None) -> AgentRedirectResponse

Parameters Return Value Example Response Types JSON Responses json(data: dict, metadata: dict = None) -> AgentResponseTypeParameters Return Value Example Text Responses text(data: str, metadata: dict = None) -> AgentResponseTypeParameters Return Value Example Binary Responses binary(data: bytes, metadata: dict = None) -> AgentResponseTypeParameters Return Value Example Media Type Responses Example HTML Responses html(data: str, metadata: dict = None) -> AgentResponseTypeParameters Return Value Example Empty Responses empty(metadata: dict = None) -> AgentResponseTypeParameters Return Value Example Request Handling Accessing Request Data trigger() -> strReturn Value Example metadata(key: str, default_value = None) -> AnyParameters Return Value Example json() -> dictReturn Value Example text() -> strReturn Value Example binary() -> bytesReturn Value Example Media-Specific Methods Example email() -> EmailReturn Value Example Email Class Email Properties subject -> strfrom_email -> str | Nonefrom_name -> str | Noneto -> str | Nonedate -> datetime | NonemessageId -> strheaders -> dicttext -> strhtml -> strattachments -> listEmail Methods to_dict() -> dictReturn Value Example Dictionary-Style Access Logging Logger Interface Logging Methods debug(message: str, *args, **kwargs) -> NoneParameters Example info(message: str, *args, **kwargs) -> NoneParameters Example warn(message: str, *args, **kwargs) -> NoneParameters Example error(message: str, *args, **kwargs) -> NoneParameters Example Creating Child Loggers child(**kwargs) -> LoggerParameters Return Value Example Telemetry Tracing Example