Overview
The Agent class provides a high-level interface for creating AI agents that can:
- Stream responses from LLM providers
- Execute tools and handle multi-turn conversations
- Support steering and follow-up messages during execution
- Manage conversation state and event subscriptions
Constructor
import { Agent } from '@mariozechner/pi-agent-core';
const agent = new Agent(options);
AgentOptions
Initial state for the agent including system prompt, model, thinking level, tools, and messages.
convertToLlm
(messages: AgentMessage[]) => Message[] | Promise<Message[]>
Converts AgentMessage[] to LLM-compatible Message[] before each LLM call. Default filters to user/assistant/toolResult messages.convertToLlm: (messages) => messages.filter(
m => m.role === 'user' || m.role === 'assistant' || m.role === 'toolResult'
)
transformContext
(messages: AgentMessage[], signal?: AbortSignal) => Promise<AgentMessage[]>
Optional transform applied to context before convertToLlm. Use for context pruning, injecting external context, etc.transformContext: async (messages) => {
if (estimateTokens(messages) > MAX_TOKENS) {
return pruneOldMessages(messages);
}
return messages;
}
steeringMode
'all' | 'one-at-a-time'
default:"one-at-a-time"
Controls how steering messages are delivered:
all: Send all steering messages at onceone-at-a-time: Send one steering message per turn
followUpMode
'all' | 'one-at-a-time'
default:"one-at-a-time"
Controls how follow-up messages are delivered:
all: Send all follow-up messages at onceone-at-a-time: Send one follow-up message per turn
Custom stream function for proxy backends or custom LLM routing. Default uses streamSimple from @mariozechner/pi-ai.
Optional session identifier forwarded to LLM providers. Used by providers that support session-based caching (e.g., OpenAI Codex).
getApiKey
(provider: string) => Promise<string | undefined> | string | undefined
Resolves an API key dynamically for each LLM call. Useful for expiring tokens (e.g., GitHub Copilot OAuth).
Custom token budgets for thinking levels (token-based providers only).
Preferred transport for providers that support multiple transports ('sse' or 'responses').
Maximum delay in milliseconds to wait for a retry when the server requests a long wait. If the server’s requested delay exceeds this value, the request fails immediately. Set to 0 to disable the cap.
Properties
state
Returns the current agent state containing:
The system prompt used for LLM calls
The LLM model to use for generation
Thinking/reasoning level: 'off' | 'minimal' | 'low' | 'medium' | 'high' | 'xhigh'
Available tools the agent can execute
Conversation history including user, assistant, and tool result messages
Whether the agent is currently streaming a response
The partial message being streamed (null when not streaming)
Set of tool call IDs currently being executed
Error message from the last failed operation
sessionId
get sessionId(): string | undefined
set sessionId(value: string | undefined)
thinkingBudgets
get thinkingBudgets(): ThinkingBudgets | undefined
set thinkingBudgets(value: ThinkingBudgets | undefined)
transport
get transport(): Transport
'sse' or 'responses').
maxRetryDelayMs
get maxRetryDelayMs(): number | undefined
set maxRetryDelayMs(value: number | undefined)
Methods
subscribe
subscribe(fn: (e: AgentEvent) => void): () => void
const unsubscribe = agent.subscribe((event) => {
switch (event.type) {
case 'message_start':
console.log('Message started:', event.message);
break;
case 'message_update':
console.log('Message updated:', event.message);
break;
case 'tool_execution_start':
console.log('Tool execution started:', event.toolName);
break;
}
});
// Later, unsubscribe
unsubscribe();
prompt
prompt(message: AgentMessage | AgentMessage[]): Promise<void>
prompt(input: string, images?: ImageContent[]): Promise<void>
// Text prompt
await agent.prompt('What is the capital of France?');
// Text with images
await agent.prompt('What is in this image?', [
{ type: 'image', source: { type: 'base64', media_type: 'image/png', data: '...' } }
]);
// Custom message
await agent.prompt({
role: 'user',
content: [{ type: 'text', text: 'Hello!' }],
timestamp: Date.now()
});
// Multiple messages
await agent.prompt([
{ role: 'user', content: 'First message', timestamp: Date.now() },
{ role: 'user', content: 'Second message', timestamp: Date.now() }
]);
Throws an error if the agent is already streaming. Use steer() or followUp() to queue messages during execution.
continue
continue(): Promise<void>
// After an error, retry the last request
await agent.continue();
steer
steer(m: AgentMessage): void
// While agent is running
agent.steer({
role: 'user',
content: 'Stop and focus on security instead',
timestamp: Date.now()
});
followUp
followUp(m: AgentMessage): void
// Queue a follow-up question
agent.followUp({
role: 'user',
content: 'Can you explain that in more detail?',
timestamp: Date.now()
});
abort
Abort the current streaming operation.
waitForIdle
waitForIdle(): Promise<void>
await agent.waitForIdle();
console.log('Agent is idle');
reset
Reset the agent to initial state. Clears messages, streaming state, pending tool calls, errors, and message queues.
State Mutators
setSystemPrompt(v: string): void
setModel(m: Model<any>): void
setThinkingLevel(l: ThinkingLevel): void
setTools(t: AgentTool<any>[]): void
setSteeringMode(mode: 'all' | 'one-at-a-time'): void
setFollowUpMode(mode: 'all' | 'one-at-a-time'): void
setTransport(value: Transport): void
replaceMessages(ms: AgentMessage[]): void
appendMessage(m: AgentMessage): void
clearMessages(): void
clearSteeringQueue(): void
clearFollowUpQueue(): void
clearAllQueues(): void
agent.setSystemPrompt('You are a helpful coding assistant.');
agent.setModel(getModel('openai', 'gpt-4o'));
agent.setThinkingLevel('medium');
agent.setTools([myTool]);
hasQueuedMessages
hasQueuedMessages(): boolean
if (agent.hasQueuedMessages()) {
console.log('Messages are queued');
}
Complete Example
import { Agent } from '@mariozechner/pi-agent-core';
import { getModel } from '@mariozechner/pi-ai';
import { Type } from '@sinclair/typebox';
// Define a tool
const weatherTool = {
label: 'Weather',
name: 'get_weather',
description: 'Get current weather for a location',
parameters: Type.Object({
location: Type.String({ description: 'City name' }),
}),
execute: async (toolCallId, args) => {
// Simulate API call
return {
content: [{
type: 'text',
text: `Weather in ${args.location}: Sunny, 72°F`
}],
details: { temperature: 72, condition: 'sunny' }
};
}
};
// Create agent
const agent = new Agent({
initialState: {
systemPrompt: 'You are a helpful weather assistant.',
model: getModel('openai', 'gpt-4o'),
thinkingLevel: 'low',
tools: [weatherTool]
}
});
// Subscribe to events
agent.subscribe((event) => {
if (event.type === 'message_update') {
console.log('Assistant:', event.message.content);
}
if (event.type === 'tool_execution_start') {
console.log(`Calling ${event.toolName}...`);
}
});
// Send prompt
await agent.prompt('What is the weather in San Francisco?');
// Access results
const lastMessage = agent.state.messages[agent.state.messages.length - 1];
console.log('Final response:', lastMessage);
Advanced: Proxy Stream Function
For applications that need to route LLM calls through a backend server:
import { streamProxy } from '@mariozechner/pi-agent-core';
const agent = new Agent({
streamFn: (model, context, options) =>
streamProxy(model, context, {
...options,
authToken: await getAuthToken(),
proxyUrl: 'https://api.example.com',
}),
});