Agentuity Docs

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.

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

welcome() -> dict

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

sendReply(request, context, subject=None, text=None, html=None, attachments=None, from_email=None, from_name=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
from_email (str, optional): Email address of the sender. Defaults to the original recipient if not provided. Can only be overridden if custom email sending is enabled
from_name (str, optional): Name of the sender. Defaults to the agent name if not provided

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 = await 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],
        from_email="support@yourcompany.com",
        from_name="Support Team"
    )
 
    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

data

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 = await 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 = await 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.

get

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 async 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 = await value.data.json()
        context.logger.info(f"User preferences: {user_prefs}")
 
        # Or access as text if needed
        # text_data = await 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"}

set

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})

delete

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.

upsert

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

Inserts or updates vectors in the vector storage.

Parameters

name: The name of the vector storage
documents: A list of documents to upsert. Each document must include a unique key and either document (text) or embeddings

Return Value

Returns a list of string IDs for the upserted vectors (internal system IDs, not your provided keys).

Note: Upserting with an existing key updates that vector (idempotent operation).

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({
            "key": product["id"],  # REQUIRED: unique key for each document
            "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",
    [
        {"key": "key_123", "document": "Ergonomic office chair with lumbar support", "metadata": {"category": "furniture"}},
        {"key": "key_456", "document": "Wireless noise-cancelling headphones", "metadata": {"category": "electronics"}}
    ]
)
 
# Upsert documents with embeddings
ids2 = await context.vector.upsert(
    "product-embeddings",
    [
        {"key": "key_123", "embeddings": [0.1, 0.2, 0.3, 0.4], "metadata": {"productId": "123"}},
        {"key": "key_456", "embeddings": [0.5, 0.6, 0.7, 0.8], "metadata": {"productId": "456"}}
    ]
)

search

async search(name: str, query: str, limit: int = 10, similarity: float = 0.5, metadata: Optional[dict] = None) -> list[VectorSearchResult]

Searches for vectors in the vector storage.

Parameters

name: The name of the vector storage
query: The query text to search against
limit: (optional) Maximum number of results to return. Defaults to 10
similarity: (optional) Minimum similarity threshold (0.0-1.0). Defaults to 0.5
metadata: (optional) Exact-match metadata filters applied at search time. Defaults to None

Return Value

Returns a list of search results with the following attributes:

id: Internal vector ID (e.g., "vector_94c98660c8afa3a2")
key: The unique key for this vector
similarity: Similarity score from 0.0 to 1.0 (1.0 = perfect match)
metadata: The associated metadata dictionary

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",
        "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.similarity}")
            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

get

async get(name: str, key: str) -> VectorSearchResult | None

Retrieves a specific vector from the vector storage using its key.

Parameters

name: The name of the vector storage
key: The unique key of the vector to retrieve

Return Value

Returns a VectorSearchResult object if found, or None if the key doesn't exist.

Example

# Retrieve a specific vector by key
vector = await context.vector.get("product-descriptions", "chair-001")
 
if vector:
    print(f"Found vector: {vector.id}")
    print(f"Key: {vector.key}")
    print(f"Metadata: {vector.metadata}")
else:
    print("Vector not found")

delete

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

Deletes a vector from the vector storage.

Parameters

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

Return Value

Returns the number of vectors that were deleted (0 if the key didn't exist, 1 if successfully deleted).

Example

from agentuity.sdk import AgentContext
 
# Delete a single vector with error handling
try:
    deleted_count = await context.vector.delete("product-descriptions", "key_123")
    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.

get

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

put

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 = b"\x00\x01..."  # example 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"
        }
    )
)

delete

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

create_public_url

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.

get_agent

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",
    "projectId": "project-456" # Note: projectId is camelCase
})
 
# Invoke the agent with error handling
try:
    # Pass raw payload (content type inferred)
    result = await agent.run("process this")
    # 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

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"},  # params
    "process this",                     # payload (inferred as text/plain)
    metadata={"source": "web-agent"}    # metadata (keyword arg)
)
# Note: Python infers content-type from data (string→text/plain, dict→application/json)

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

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

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

async json() -> dict

Gets the payload of the request as a JSON object.

Return Value

Returns the request payload as a dictionary.

Example

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

text

async text() -> str

Gets the payload of the request as a string.

Return Value

Returns the request payload as a string.

Example

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

binary

async binary() -> bytes

Gets the payload of the request as bytes.

Return Value

Returns the request payload as bytes.

Example

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

Media-Specific Methods

The SDK provides specialized methods for various media types:

async pdf() -> bytes
async png() -> bytes
async jpeg() -> bytes
async gif() -> bytes
async webp() -> bytes
async mp3() -> bytes
async wav() -> bytes
async ogg() -> bytes
async 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 = await request.data.png()
# Process the image...
 
# Get audio from the request
audio = await request.data.mp3()
# Process the audio...

email

async 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

subject -> str

Returns the subject line of the email.

from_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

from_name -> str | None

Returns the sender's display name if available.

to

to -> str | None

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

date

date -> datetime | None

Returns the email date as a datetime object.

messageId

messageId -> str

Returns the unique message ID of the email.

headers

headers -> dict

Returns all email headers as a dictionary.

text

text -> str

Returns the plain text content of the email.

html

html -> str

Returns the HTML content of the email.

attachments

attachments -> list

Returns a list of email attachments.

Email Methods

to_dict

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 Python SDK provides a standard logging.Logger instance through context.logger:

class Logger(logging.Logger):
    """
    Logger for agent execution.
    """
    def debug(self, message: str, *args) -> None:
        """Log a debug message."""
        pass
 
    def info(self, message: str, *args) -> None:
        """Log an informational message."""
        pass
 
    def warning(self, message: str, *args) -> None:
        """Log a warning message."""
        pass
 
    def error(self, message: str, *args) -> None:
        """Log an error message."""
        pass
 
    # Note: child() method coming soon for API consistency with JavaScript SDK

Logging Methods

debug

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

Logs a debug message.

Parameters

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

Example

request_id = "123"
context.logger.debug(f"Processing request request_id={request_id}")

info

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

Logs an informational message.

Parameters

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

Example

request_id = "123"
context.logger.info(f"Request processed successfully request_id={request_id}")

warning

warning(message: str, *args) -> None

Logs a warning message.

Parameters

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

Example

resource_id = "456"
context.logger.warning(f"Resource not found resource_id={resource_id}")

error

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

Logs an error message.

Parameters

message: The message to log
args: 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(f"Failed to process request: {str(e)}")

Creating Child Loggers

Child logger functionality is coming soon for API consistency with the JavaScript SDK.

Current approach (manual context):

# Python: Include context in each log message  
request_id = "123"
user_id = "456"
endpoint = "/api/users"
 
context.logger.info(f"Starting request processing request_id={request_id} endpoint={endpoint}")
context.logger.info(f"User authenticated request_id={request_id} user_id={user_id}")
context.logger.error(f"Failed to fetch user data request_id={request_id} error={str(error)}")

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!

API Reference

Need Help?

On this page