Skip to content
This repository was archived by the owner on Mar 25, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 89 additions & 0 deletions content/docs/SDKs/python/agent-to-agent.mdx
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
---
title: Agent-to-Agent Communication
description: Enable communication between agents in the Agentuity Python SDK
---

# Agent-to-Agent Communication

The Agentuity Python SDK enables seamless communication between agents, whether they are running locally or in different environments.

## Local Agent Communication

You can easily communicate between agents running in the same process:

```python
async def run(request, response, context):
# Call another agent by ID
return await response.handoff({"id": "other_agent_id"})

# Or call by name
# return response.handoff({"name": "OtherAgentName"})
```

## Remote Agent Communication

The SDK now supports communication with agents running in different environments:

```python
async def run(request, response, context):
# Resolve an agent (local or remote)
agent = context.get_agent({"id": "remote_agent_id"})

# Create data for the remote agent
from agentuity.server.data import Data, StringStreamReader
data = Data("text/plain", StringStreamReader("Hello from another agent"))

# Call the remote agent
result = await agent.run(data, metadata={"key": "value"})

# Process the response
text = await result.data.text()
return response.text(f"Got response: {text}")
```

## Agent Resolution

The `AgentContext` provides methods for resolving agents, whether they are local or remote:

```python
async def run(request, response, context):
# Resolve by ID
agent = context.get_agent({"id": "agent_id"})

# Resolve by name
agent = context.get_agent({"name": "AgentName"})

# Resolve by name in a specific project
agent = context.get_agent({"name": "AgentName", "projectId": "proj_123"})

# Call the resolved agent
result = await agent.run(request.data)
return result
```

## Agent Context

The `AgentContext` class provides methods for agent resolution and management:

```python
async def run(request, response, context):
# Get agent information
current_agent = context.agent
all_agents = context.agents

# Get the current scope (local or remote)
scope = context.scope

# Get the current run ID
run_id = context.runId

# Use key-value store
await context.kv.set("key", "value")
value = await context.kv.get("key")

# Use vector store
await context.vector.add("collection", "id", [0.1, 0.2, 0.3], {"metadata": "value"})
results = await context.vector.search("collection", [0.1, 0.2, 0.3], 5)

return response.text("Agent context example")
```
86 changes: 86 additions & 0 deletions content/docs/SDKs/python/async-api.mdx
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
---
title: Async API
description: Working with Asynchronous Agents in the Agentuity Python SDK
---

# Async API

The Agentuity Python SDK now supports fully asynchronous operations, enabling more efficient handling of concurrent requests and streaming responses.

## Agent Request

The `AgentRequest` class now uses async patterns to handle incoming data:

```python
from agentuity.server import AgentRequest

async def run(request: AgentRequest, response, context):
# Access request data asynchronously
data = await request.data.text()

# Access request metadata
metadata = request.metadata

# Get the trigger type
trigger = request.trigger

# Return a response
return response.text(f"Received: {data}")
```

## Agent Response

The `AgentResponse` class supports async streaming and iteration:

```python
from agentuity.server import AgentResponse

async def run(request, response: AgentResponse, context):
# Stream response data
async def data_generator():
for i in range(5):
yield f"Data chunk {i}\n"

# Return a streaming response
return response.stream(data_generator(), contentType="text/plain")
```

### Async Iteration

Responses can be iterated over asynchronously:

```python
async def process_response(response):
async for chunk in response:
print(chunk)
```

## Data Streaming

The SDK provides several specialized stream readers for different data types:

```python
from agentuity.server.data import (
StringStreamReader,
BytesStreamReader,
EmptyDataReader
)

# Create a stream from a string
string_stream = StringStreamReader("Hello, world!")

# Create a stream from bytes
bytes_stream = BytesStreamReader(b"Binary data")

# Create an empty stream
empty_stream = EmptyDataReader()
```

## Performance Benefits

The async implementation provides several advantages:

- Efficient handling of concurrent requests
- Streaming of large data payloads without blocking
- Reduced memory usage for processing large files
- Better integration with modern async Python applications
118 changes: 118 additions & 0 deletions content/docs/SDKs/python/data-handling.mdx
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
---
title: Data Handling
description: Working with various data types in the Agentuity Python SDK
---

# Data Handling

The Agentuity Python SDK provides a flexible system for working with different types of data, from text and JSON to binary formats like images and audio.

## The Data Class

The `Data` class serves as a container for all types of content:

```python
from agentuity.server.data import Data, StringStreamReader, BytesStreamReader

# Create a text Data object
text_data = Data("text/plain", StringStreamReader("Hello, world!"))

# Create a binary Data object
binary_data = Data("application/octet-stream", BytesStreamReader(b"\x00\x01\x02\x03"))
```

## Accessing Data Content

Data content can be accessed in various formats:

```python
async def process_data(data: Data):
# Get as text
text = await data.text()

# Get as JSON
json_data = await data.json()

# Get as binary
binary = await data.binary()

# Get as base64 encoded string
base64_str = await data.base64()

# Access as a stream
stream = await data.stream()
async for chunk in stream:
process_chunk(chunk)
```

## Working with Responses

The `AgentResponse` class provides methods for various content types:

```python
async def run(request, response, context):
# Text response
return response.text("Hello, world!")

# JSON response
return response.json({"message": "Hello, world!"})

# Markdown response
return response.markdown("# Hello, world!")

# Binary response
return response.binary(b"\x00\x01\x02\x03", "application/octet-stream")
```

## Media Type Helpers

The SDK includes helpers for common media types:

```python
async def run(request, response, context):
# Image responses
with open("image.png", "rb") as f:
image_data = f.read()
return response.png(image_data)

# Other image formats
return response.jpeg(jpeg_data)
return response.gif(gif_data)
return response.webp(webp_data)

# Document formats
return response.pdf(pdf_data)

# Audio/Video formats
return response.mp3(mp3_data)
return response.mp4(mp4_data)
return response.wav(wav_data)
return response.ogg(ogg_data)
return response.webm(webm_data)
return response.m4a(m4a_data)
```

## Streaming Data

For large responses or real-time data, use streaming:

```python
async def run(request, response, context):
# Stream data from a generator
async def generator():
for i in range(10):
yield f"Item {i}\n"

# Return the stream
return response.stream(generator(), contentType="text/plain")

# Transform items during streaming
def transform(item):
return f"Transformed: {item}"

return response.stream(generator(), transform=transform)

# Stream responses from another agent
other_response = await context.get_agent("other_agent").run(request.data)
return response.stream(other_response)
```
75 changes: 75 additions & 0 deletions content/docs/SDKs/python/index.mdx
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
---
title: Python SDK
description: Agentuity Python SDK Documentation
---

# Agentuity Python SDK

The Agentuity Python SDK is a powerful toolkit for building, deploying, and managing AI agents in Python environments. This SDK provides developers with a comprehensive set of tools to create intelligent agents that can process various types of content, communicate with each other, and integrate with external systems.

## Prerequisites

- [Python](https://www.python.org/) (3.10 or 3.11)
- [uv](https://docs.astral.sh/uv/) (latest version recommended)

## Installation

First, install the Agentuity CLI:

```bash
curl -sSL https://agentuity.sh/install.sh | bash
```

Then create a new project:

```bash
agentuity new
```

This will create a new project with a default `agentuity.yaml` configuration file and agent structure.

## Creating Your First Agent

Navigate to the agent file created by the CLI:

```bash
cd agents/myfirstagent
```

Edit the `agent.py` file and add the following code:

```python
from agentuity.server import AgentRequest, AgentResponse, AgentContext

def run(request: AgentRequest, response: AgentResponse, context: AgentContext):
return response.text("Hello, world")
```

## Running Your Agent

Start the local development server:

```bash
agentuity dev
```

When the server starts, you'll see a local URL in the console output. Open this URL in your browser to access the Agentuity development UI, where you can test your agent and view the responses.

## Key Features

<div style={{ display: 'grid', gridTemplateColumns: 'repeat(auto-fit, minmax(250px, 1fr))', gap: '1rem', padding: '1rem' }}>
<Card href="/docs/SDKs/python/async-api">
<h3>Async API</h3>
<p>Work with fully asynchronous agents</p>
</Card>

<Card href="/docs/SDKs/python/agent-to-agent">
<h3>Agent-to-Agent Communication</h3>
<p>Enable communication between local and remote agents</p>
</Card>

<Card href="/docs/SDKs/python/data-handling">
<h3>Data Handling</h3>
<p>Work with various data types and streaming capabilities</p>
</Card>
</div>