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
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-tokencontext
(AgentContext): The agent context, used to get the base_url and agentIdsubject
(str, optional): Subject of the reply. Defaults to 'Re: <original subject>'text
(str, optional): Plain text body of the replyhtml
(str, optional): HTML body of the replyattachments
(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 attachmentdata
(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 attachmentdata()
: 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 attachmentcontent_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
: AnAgentRequest
object containing the request dataresponse
: AnAgentResponse
object for creating responsescontext
: AnAgentContext
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 storagekey
: 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 storagekey
: The key to store the value undervalue
: The value to store (Union[str, int, float, bool, list, dict, bytes, "Data"])params
(optional): Can havettl
(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 storagekey
: 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 storagedocuments
: 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 storageparams
: 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 storagekey
: 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 fromkey
: 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 intokey
: The key of the object to putdata
: 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), andcontent_language
(Content-Language). The optionalmetadata
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. Thecontent_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 toapplication/json
). If the content type cannot be determined, it defaults toapplication/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 fromkey
: 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 objectkey
: The key of the objectexpires_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 agentpayload
(optional): The payload to send to the target agentcontent_type
(optional): The content type of the payloadmetadata
(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 responsemetadata
(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 responsemetadata
(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 responsemetadata
(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 responsemetadata
(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 retrievedefault_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 logargs
,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 logargs
,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 logargs
,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 logargs
,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!