The Agent Development Kit (ADK) provides modules for agents running in the agent runtime. These modules handle communication with the platform, state management, filesystem operations, and more.
Overview
Import ADK modules from terminaluse.lib:
from terminaluse.lib import adk
Available Modules
| Module | Purpose |
|---|
adk.messages | Send and list messages |
adk.state | Create, get, update, delete task state |
adk.tasks | Get and delete tasks |
adk.events | List and get events |
adk.agents | Retrieve agent information |
adk.acp | Agent-to-Client Protocol (create tasks, send events) |
adk.filesystem | Sync files up/down with change detection |
adk.agent_task_tracker | Track agent task progress |
adk.task | Sync task-scoped system folders |
adk.messages
Send and manage messages within a task.
send
Send a message to the task.
from terminaluse.lib import adk
from terminaluse.lib import TextPart, DataPart
# Simple text message (recommended)
await adk.messages.send(
task_id="task-123",
content="Hello, world!"
)
# Explicit TextPart
await adk.messages.send(
task_id="task-123",
content=TextPart(text="**Bold** and *italic*")
)
# Structured data (for generative UIs)
await adk.messages.send(
task_id="task-123",
content=DataPart(data={"temperature": 72, "unit": "fahrenheit"})
)
list
List messages for a task.
messages = await adk.messages.list(task_id="task-123")
for msg in messages:
print(msg.content)
adk.state
Manage persistent state for a task. State is scoped to (task_id, agent_id).
create
Initialize state when a task starts.
await adk.state.create(
task_id="task-123",
agent_id="agent-456",
state={
"user_preferences": {},
"message_count": 0,
"initialized_at": "2024-01-01T00:00:00Z"
}
)
get
Retrieve current state.
state = await adk.state.get(
task_id="task-123",
agent_id="agent-456"
)
print(state.get("message_count"))
update
Merge updates into existing state.
await adk.state.update(
task_id="task-123",
agent_id="agent-456",
state={
"message_count": 5,
"last_activity": "2024-01-01T12:00:00Z"
}
)
delete
Remove state entirely.
await adk.state.delete(
task_id="task-123",
agent_id="agent-456"
)
adk.tasks
Retrieve and manage tasks.
get
Get a task by ID or name.
# By ID
task = await adk.tasks.get(task_id="task-123")
# By name
task = await adk.tasks.get(task_name="my-task")
print(task.id, task.status)
delete
Delete a task.
await adk.tasks.delete(task_id="task-123")
adk.events
Access events for a task.
list
List events for a task.
events = await adk.events.list(task_id="task-123")
for event in events:
print(event.content)
get
Get a specific event.
event = await adk.events.get(event_id="event-789")
adk.agents
Retrieve agent information.
get
Get an agent by ID or name.
# By ID
agent = await adk.agents.get(agent_id="agent-456")
# By name
agent = await adk.agents.get(agent_name="my-agent")
print(agent.name, agent.status)
list
List agents.
agents = await adk.agents.list()
for agent in agents:
print(agent.name)
adk.acp
Agent-to-Client Protocol for creating tasks and sending events programmatically.
create_task
Create a new task.
task = await adk.acp.create_task(
agent_id="agent-456",
params={"user_id": "user-123", "mode": "interactive"}
)
print(task.id)
send_event
Send an event to a task.
await adk.acp.send_event(
task_id="task-123",
content=TextPart(text="User input")
)
cancel_task
Cancel a running task.
await adk.acp.cancel_task(task_id="task-123")
adk.filesystem
Sync files between the agent runtime and cloud storage.
sync_down
Download files from cloud storage to local filesystem.
result = await adk.filesystem.sync_down(
filesystem_id="fs-123",
local_path="/workspace/files"
)
print(f"Downloaded {result.files_changed} files")
sync_up
Upload local files to cloud storage.
result = await adk.filesystem.sync_up(
filesystem_id="fs-123",
local_path="/workspace/files"
)
print(f"Uploaded {result.files_changed} files")
- Manifest-based sync: Only changed files are transferred
- Compression: Files are compressed for efficient transfer
- Change detection: Automatically detects modified files
See Filesystem Sync for detailed usage.
adk.task
Task-scoped system-folder helpers.
sync_down_system_folder
await adk.task.sync_down_system_folder(
task_id="task-123",
folder_type="dot_claude",
)
sync_up_system_folder
await adk.task.sync_up_system_folder(
task_id="task-123",
folder_type="dot_claude",
)
/root/.claude or /root/.codex separately from /workspace.
adk.agent_task_tracker
Track progress of agent tasks.
get
Get tracker for a task.
tracker = await adk.agent_task_tracker.get(task_id="task-123")
print(tracker.status, tracker.progress)
list
List all trackers.
trackers = await adk.agent_task_tracker.list()
update
Update tracker progress.
await adk.agent_task_tracker.update(
task_id="task-123",
progress=75,
status_message="Processing step 3 of 4"
)
Using with TaskContext
In agent handlers, use TaskContext for convenience - it provides pre-bound versions of these modules:
from terminaluse.lib import AgentServer, TaskContext, Event
server = AgentServer()
@server.on_event
async def handle_event(ctx: TaskContext, event: Event):
# Using ctx (pre-bound to current task/agent)
await ctx.messages.send("Hello")
await ctx.state.update({"count": 1})
# Equivalent using adk (requires explicit IDs)
from terminaluse.lib import adk
await adk.messages.send(task_id=ctx.task.id, content="Hello")
await adk.state.update(
task_id=ctx.task.id,
agent_id=ctx.agent.id,
state={"count": 1}
)
Use ctx in handlers for cleaner code. Use adk directly in helper functions or when you need to operate on different tasks.
