API Reference

This section contains the API documentation for Agenix, automatically generated from the source code.

Package Overview

Agenix - A lightweight AI coding agent.

A minimal Python implementation inspired by pi-mono, supporting OpenAI and Anthropic LLMs with file operations, shell execution, Agent Skills, and Extensions.

## Programmatic Usage (SDK)

```python import asyncio from agenix import create_session

async def main():

session = await create_session(: api_key=”your-key”, model=”gpt-4o”

) response = await session.prompt(“What files are in the current directory?”) print(response)

asyncio.run(main()) ```

## Extension System

Extensions can be placed in: - Global: ~/.agenix/extensions/ - Project: .agenix/extensions/

Example extension: ```python # ~/.agenix/extensions/my_tool.py async def setup(agenix):

from agenix.extensions import ToolDefinition, EventType

async def my_execute(params, ctx):
return f”Result: {params}”

agenix.register_tool(ToolDefinition(
name=”my_tool”, description=”My custom tool”, parameters={“type”: “object”}, execute=my_execute

))

```

async agenix.create_session(api_key: str | None = None, base_url: str | None = None, model: str = 'gpt-4o', system_prompt: str | None = None, working_dir: str = '.', max_turns: int = 10, enable_extensions: bool = True) → AgentSession[source]

Create an agent session for programmatic usage.

Parameters:

api_key – API key for the LLM provider (or set OPENAI_API_KEY env var)
base_url – Optional API base URL
model – Model to use (default: gpt-4o)
system_prompt – Optional custom system prompt
working_dir – Working directory for file operations
max_turns – Maximum conversation turns per prompt
enable_extensions – Whether to load extensions

Returns:

AgentSession instance

Example

>>> import asyncio
>>> from agenix import create_session
>>>
>>> async def main():
...     session = await create_session(
...         api_key="your-key",
...         model="gpt-4o"
...     )
...     response = await session.prompt("Hello!")
...     print(response)
...
>>> asyncio.run(main())

class agenix.AgentSession(agent: Agent, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

High-level agent session for programmatic usage.

Wraps the low-level Agent class with a simpler API.

__init__(agent: Agent, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

async prompt(message: str) → str[source]

Send a prompt to the agent and return the response.

Parameters:: message – User message/prompt
Returns:: Agent’s text response

get_messages() → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]

Get conversation history.

Returns:: List of messages in the conversation

clear_messages() → None[source]: Clear conversation history.

async close() → None[source]: Close the session and cleanup.

class agenix.Agent(config: AgentConfig, tools: List[Tool] | None = None)[source]

Agent runtime with tool execution loop.

__init__(config: AgentConfig, tools: List[Tool] | None = None)[source]

Subscribe to agent events.

Returns:: Unsubscribe function

Process a user prompt through the agent loop.

Parameters:: user_message – User’s message
Yields:: Agent events

get_messages() → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]: Get conversation messages.

clear_messages() → None[source]: Clear conversation history.

class agenix.AgentConfig(model: str, api_key: str, base_url: str | None = None, system_prompt: str | None = None, max_turns: int = 10, max_tool_calls_per_turn: int = 20, max_tokens: int = 16384, skill_dirs: List[str] | None = None)[source]

Agent configuration.

model: str

api_key: str

base_url: str | None = None

system_prompt: str | None = None

max_turns: int = 10

max_tool_calls_per_turn: int = 20

max_tokens: int = 16384

skill_dirs: List[str] | None = None

__post_init__()[source]: Create provider after initialization.

__init__(model: str, api_key: str, base_url: str | None = None, system_prompt: str | None = None, max_turns: int = 10, max_tool_calls_per_turn: int = 20, max_tokens: int = 16384, skill_dirs: List[str] | None = None) → None

agenix.get_provider(provider_name: str, **kwargs) → LLMProvider[source]: Get LLM provider by name.

class agenix.SessionManager(session_dir: str = '.agenix')[source]

Manage persistent agent sessions.

__init__(session_dir: str = '.agenix')[source]

create_session(name: str | None = None) → str[source]

Create a new session.

Parameters:: name – Optional session name
Returns:: Session ID

save_message(session_id: str, message: UserMessage | AssistantMessage | ToolResultMessage | SystemMessage) → None[source]

Save a message to session.

Parameters:

session_id – Session ID
message – Message to save

load_session(session_id: str) → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]

Load session messages.

Parameters:: session_id – Session ID
Returns:: List of messages

list_sessions() → List[Dict[str, Any]][source]

List all sessions.

Returns:: List of session metadata

delete_session(session_id: str) → None[source]

Delete a session.

Parameters:: session_id – Session ID

class agenix.Skill(name: str, description: str, file_path: str, content: str, license: str | None = None, compatibility: str | None = None, allowed_tools: List[str] | None = None, metadata: Dict[str, ~typing.Any]=<factory>, disable_model_invocation: bool = False)[source]

Represents a skill with its metadata and content.

A skill extends Claude’s capabilities by providing specialized knowledge or workflows. Skills follow the Anthropic Agent Skills standard.

name

Unique skill identifier (lowercase, hyphens, 1-64 chars)

Type:: str

description

Brief description of the skill’s purpose

Type:: str

file_path

Absolute path to the skill’s markdown file

Type:: str

content

Full markdown content including frontmatter

Type:: str

license

License terms for the skill

Type:: Optional[str]

compatibility

Compatibility requirements (e.g., “Python 3.8+”)

Type:: Optional[str]

allowed_tools

List of tools this skill can use

Type:: Optional[List[str]]

metadata

Additional custom metadata

Type:: Dict[str, Any]

disable_model_invocation

If True, skill is not shown to model

Type:: bool

Examples

>>> skill = Skill(
...     name="pdf",
...     description="PDF manipulation toolkit",
...     file_path="/skills/pdf/SKILL.md",
...     content="---\nname: pdf\n...",
...     allowed_tools=["Bash(pdf:*)"]
... )
>>> print(skill.name)
'pdf'

name: str

description: str

file_path: str

content: str

license: str | None = None

compatibility: str | None = None

allowed_tools: List[str] | None = None

metadata: Dict[str, Any]

disable_model_invocation: bool = False

__init__(name: str, description: str, file_path: str, content: str, license: str | None = None, compatibility: str | None = None, allowed_tools: List[str] | None = None, metadata: Dict[str, ~typing.Any]=<factory>, disable_model_invocation: bool = False) → None

class agenix.SkillManager(skill_dirs: List[str] | None = None)[source]

Manages discovery, loading, and access to skills.

The SkillManager discovers skills from configured directories, parses their frontmatter, validates them according to the Agent Skills standard, and provides access to skill metadata and content.

Skill discovery looks for: - Direct .md files in the root of skill directories - SKILL.md files in subdirectories (subdirectory name should match skill name)

Default skill directories (if no custom dirs specified): - Global: ~/.agenix/skills/ - Project: .agenix/skills/

skill_dirs

List of directories to search for skills

Type:: List[str]

skills

Dictionary mapping skill names to Skill objects

Type:: Dict[str, Skill]

Examples

>>> # Use default directories
>>> manager = SkillManager()
>>> print(len(manager.list_skills()))
5

>>> # Custom directories
>>> manager = SkillManager(skill_dirs=['/my/skills', '/other/skills'])
>>> skill = manager.get_skill('pdf')
>>> content = manager.load_skill_content('pdf')

>>> # Generate summary for system prompt
>>> summary = manager.get_skills_summary()
>>> print(summary)
<available_skills>
  <skill name="pdf">
    PDF manipulation toolkit...
  </skill>
</available_skills>

__init__(skill_dirs: List[str] | None = None)[source]

Initialize skill manager and load skills.

Parameters:: skill_dirs – List of directories to search for skills. If None, uses default directories (~/.agenix/skills/ and .agenix/skills/)

Examples

>>> manager = SkillManager()  # Use defaults
>>> manager = SkillManager(skill_dirs=['/custom/path'])  # Custom

skills: Dict[str, Skill]

get_skill(name: str) → Skill | None[source]

Get a skill by name.

Parameters:: name – Skill name
Returns:: Skill object if found, None otherwise

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> skill = manager.get_skill('pdf')
>>> if skill:
...     print(skill.description)

list_skills() → List[Skill][source]

List all available skills.

Returns:: List of all loaded Skill objects

Examples

>>> manager = SkillManager()
>>> for skill in manager.list_skills():
...     print(f"{skill.name}: {skill.description}")

get_skills_summary() → str[source]

Get a summary of available skills for system prompt.

Generates an XML-formatted summary of skills that are not disabled. This summary is injected into the system prompt to make Claude aware of available skills.

Returns:: XML-formatted string with skill names and descriptions, or empty string if no visible skills

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> summary = manager.get_skills_summary()
>>> print(summary)
<available_skills>
  <skill name="pdf">
    PDF manipulation toolkit...
  </skill>
</available_skills>

Note

Skills with disable_model_invocation=True are excluded from the summary. These skills exist but are not shown to the model.

load_skill_content(name: str) → str | None[source]

Load full content of a skill.

Returns the complete markdown content including frontmatter. This is used for progressive disclosure - only loading full skill content when the model needs it.

Parameters:: name – Skill name
Returns:: Full skill markdown content, or None if skill not found

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> content = manager.load_skill_content('pdf')
>>> if content:
...     print("Loaded skill documentation")

add_skill_dir(directory: str)[source]

Add a skill directory and load skills from it.

Dynamically adds a new skill directory to search and discovers any skills in that directory.

Parameters:: directory – Path to skill directory

Examples

>>> manager = SkillManager()
>>> manager.add_skill_dir('/custom/skills')
>>> print(len(manager.list_skills()))

class agenix.UserMessage(role: Literal['user'] = 'user', content: str | List[TextContent | ImageContent] = '', timestamp: float = <factory>)[source]

User message.

role: Literal['user'] = 'user'

content: str | List[TextContent | ImageContent] = ''

timestamp: float

__init__(role: Literal['user'] = 'user', content: str | List[TextContent | ImageContent] = '', timestamp: float = <factory>) → None

class agenix.AssistantMessage(role: Literal['assistant'] = 'assistant', content: str | List[TextContent | ToolCall] = '', tool_calls: List[ToolCall] = <factory>, model: str = '', usage: Usage | None = None, stop_reason: str | None = None, timestamp: float = <factory>)[source]

Assistant message with tool calls and metadata.

role: Literal['assistant'] = 'assistant'

content: str | List[TextContent | ToolCall] = ''

tool_calls: List[ToolCall]

model: str = ''

usage: Usage | None = None

stop_reason: str | None = None

timestamp: float

__init__(role: Literal['assistant'] = 'assistant', content: str | List[TextContent | ToolCall] = '', tool_calls: List[ToolCall] = <factory>, model: str = '', usage: Usage | None = None, stop_reason: str | None = None, timestamp: float = <factory>) → None

class agenix.ToolResultMessage(role: Literal['tool'] = 'tool', tool_call_id: str = '', name: str = '', content: str | List[TextContent | ImageContent] = '', is_error: bool = False, timestamp: float = <factory>)[source]

Tool execution result.

role: Literal['tool'] = 'tool'

tool_call_id: str = ''

name: str = ''

content: str | List[TextContent | ImageContent] = ''

is_error: bool = False

timestamp: float

__init__(role: Literal['tool'] = 'tool', tool_call_id: str = '', name: str = '', content: str | List[TextContent | ImageContent] = '', is_error: bool = False, timestamp: float = <factory>) → None

class agenix.TextContent(type: Literal['text'] = 'text', text: str = '')[source]

Text content block.

type: Literal['text'] = 'text'

text: str = ''

__init__(type: Literal['text'] = 'text', text: str = '') → None

class agenix.ImageContent(type: Literal['image'] = 'image', source: Dict[str, ~typing.Any]=<factory>)[source]

Image content block.

type: Literal['image'] = 'image'

source: Dict[str, Any]

__init__(type: Literal['image'] = 'image', source: Dict[str, ~typing.Any]=<factory>) → None

class agenix.ToolCall(id: str, name: str, arguments: Dict[str, Any])[source]

Tool call from assistant.

id: str

name: str

arguments: Dict[str, Any]

__init__(id: str, name: str, arguments: Dict[str, Any]) → None

class agenix.ReadTool(working_dir: str = '.')[source]

Read file content with image support.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute read operation.

class agenix.WriteTool(working_dir: str = '.')[source]

Write content to a file.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute write operation.

class agenix.EditTool(working_dir: str = '.')[source]

Edit file by replacing exact strings.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute edit operation.

class agenix.BashTool(working_dir: str = '.', timeout: int = 120)[source]

Execute bash commands.

__init__(working_dir: str = '.', timeout: int = 120)[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute bash command.

class agenix.GrepTool(working_dir: str = '.')[source]

Search for patterns in files.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute grep operation.

Core Modules

Agent

Agent runtime with tool execution loop.

class agenix.core.agent.AgentConfig(model: str, api_key: str, base_url: str | None = None, system_prompt: str | None = None, max_turns: int = 10, max_tool_calls_per_turn: int = 20, max_tokens: int = 16384, skill_dirs: List[str] | None = None)[source]

Bases: object

Agent configuration.

model: str

api_key: str

base_url: str | None = None

system_prompt: str | None = None

max_turns: int = 10

max_tool_calls_per_turn: int = 20

max_tokens: int = 16384

skill_dirs: List[str] | None = None

__post_init__()[source]: Create provider after initialization.

__init__(model: str, api_key: str, base_url: str | None = None, system_prompt: str | None = None, max_turns: int = 10, max_tool_calls_per_turn: int = 20, max_tokens: int = 16384, skill_dirs: List[str] | None = None) → None

class agenix.core.agent.Agent(config: AgentConfig, tools: List[Tool] | None = None)[source]

Bases: object

Agent runtime with tool execution loop.

__init__(config: AgentConfig, tools: List[Tool] | None = None)[source]

messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage]

subscribers: List[Callable[[AgentStartEvent | AgentEndEvent | TurnStartEvent | TurnEndEvent | MessageStartEvent | MessageUpdateEvent | MessageEndEvent | ToolExecutionStartEvent | ToolExecutionUpdateEvent | ToolExecutionEndEvent], None]]

Subscribe to agent events.

Returns:: Unsubscribe function

Process a user prompt through the agent loop.

Parameters:: user_message – User’s message
Yields:: Agent events

get_messages() → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]: Get conversation messages.

clear_messages() → None[source]: Clear conversation history.

LLM Provider

Unified LLM interface supporting multiple providers.

class agenix.core.llm.StreamEvent(type: str, delta: str = '', tool_call: ToolCall | None = None, finish_reason: str | None = None)[source]

Bases: object

LLM stream event.

type: str

delta: str = ''

tool_call: ToolCall | None = None

finish_reason: str | None = None

__init__(type: str, delta: str = '', tool_call: ToolCall | None = None, finish_reason: str | None = None) → None

class agenix.core.llm.LLMProvider(api_key: str | None = None, base_url: str | None = None)[source]

Bases: ABC

Abstract LLM provider interface.

__init__(api_key: str | None = None, base_url: str | None = None)[source]

abstractmethod async stream(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AsyncIterator[StreamEvent][source]: Stream LLM responses.

abstractmethod async complete(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AssistantMessage[source]: Complete LLM request (non-streaming).

class agenix.core.llm.OpenAIProvider(api_key: str | None = None, base_url: str | None = None)[source]

Bases: LLMProvider

OpenAI/compatible API provider.

__init__(api_key: str | None = None, base_url: str | None = None)[source]

async stream(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AsyncIterator[StreamEvent][source]: Stream OpenAI responses.

async complete(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AssistantMessage[source]: Complete OpenAI request.

class agenix.core.llm.AnthropicProvider(api_key: str | None = None, base_url: str | None = None)[source]

Bases: LLMProvider

Anthropic Claude API provider.

__init__(api_key: str | None = None, base_url: str | None = None)[source]

async stream(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AsyncIterator[StreamEvent][source]: Stream Anthropic responses.

async complete(model: str, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage], system_prompt: str | None = None, tools: List[Dict[str, Any]] | None = None, max_tokens: int = 4096) → AssistantMessage[source]: Complete Anthropic request.

agenix.core.llm.get_provider(provider_name: str, **kwargs) → LLMProvider[source]: Get LLM provider by name.

Messages

Message types for agent communication.

class agenix.core.messages.TextContent(type: Literal['text'] = 'text', text: str = '')[source]

Bases: object

Text content block.

type: Literal['text'] = 'text'

text: str = ''

__init__(type: Literal['text'] = 'text', text: str = '') → None

class agenix.core.messages.ImageContent(type: Literal['image'] = 'image', source: Dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Image content block.

type: Literal['image'] = 'image'

source: Dict[str, Any]

__init__(type: Literal['image'] = 'image', source: Dict[str, ~typing.Any]=<factory>) → None

class agenix.core.messages.ToolCall(id: str, name: str, arguments: Dict[str, Any])[source]

Bases: object

Tool call from assistant.

id: str

name: str

arguments: Dict[str, Any]

__init__(id: str, name: str, arguments: Dict[str, Any]) → None

class agenix.core.messages.Usage(input_tokens: int = 0, output_tokens: int = 0, cache_read_tokens: int = 0, cache_write_tokens: int = 0, total_cost: float = 0.0)[source]

Bases: object

Token usage statistics.

input_tokens: int = 0

output_tokens: int = 0

cache_read_tokens: int = 0

cache_write_tokens: int = 0

total_cost: float = 0.0

__init__(input_tokens: int = 0, output_tokens: int = 0, cache_read_tokens: int = 0, cache_write_tokens: int = 0, total_cost: float = 0.0) → None

class agenix.core.messages.UserMessage(role: Literal['user'] = 'user', content: str | List[TextContent | ImageContent] = '', timestamp: float = <factory>)[source]

Bases: object

User message.

role: Literal['user'] = 'user'

content: str | List[TextContent | ImageContent] = ''

timestamp: float

__init__(role: Literal['user'] = 'user', content: str | List[TextContent | ImageContent] = '', timestamp: float = <factory>) → None

class agenix.core.messages.AssistantMessage(role: Literal['assistant'] = 'assistant', content: str | List[TextContent | ToolCall] = '', tool_calls: List[ToolCall] = <factory>, model: str = '', usage: Usage | None = None, stop_reason: str | None = None, timestamp: float = <factory>)[source]

Bases: object

Assistant message with tool calls and metadata.

role: Literal['assistant'] = 'assistant'

content: str | List[TextContent | ToolCall] = ''

tool_calls: List[ToolCall]

model: str = ''

usage: Usage | None = None

stop_reason: str | None = None

timestamp: float

__init__(role: Literal['assistant'] = 'assistant', content: str | List[TextContent | ToolCall] = '', tool_calls: List[ToolCall] = <factory>, model: str = '', usage: Usage | None = None, stop_reason: str | None = None, timestamp: float = <factory>) → None

class agenix.core.messages.ToolResultMessage(role: Literal['tool'] = 'tool', tool_call_id: str = '', name: str = '', content: str | List[TextContent | ImageContent] = '', is_error: bool = False, timestamp: float = <factory>)[source]

Bases: object

Tool execution result.

role: Literal['tool'] = 'tool'

tool_call_id: str = ''

name: str = ''

content: str | List[TextContent | ImageContent] = ''

is_error: bool = False

timestamp: float

__init__(role: Literal['tool'] = 'tool', tool_call_id: str = '', name: str = '', content: str | List[TextContent | ImageContent] = '', is_error: bool = False, timestamp: float = <factory>) → None

class agenix.core.messages.SystemMessage(role: Literal['system'] = 'system', content: str = '')[source]

Bases: object

System message (for context).

role: Literal['system'] = 'system'

content: str = ''

__init__(role: Literal['system'] = 'system', content: str = '') → None

class agenix.core.messages.Context(system_prompt: str | None = None, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage] = <factory>, tools: Dict[str, ~typing.Any]]=<factory>)[source]

Bases: object

Conversation context.

system_prompt: str | None = None

messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage]

tools: List[Dict[str, Any]]

__init__(system_prompt: str | None = None, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage] = <factory>, tools: Dict[str, ~typing.Any]]=<factory>) → None

class agenix.core.messages.AgentEvent(type: str, timestamp: float = <factory>)[source]

Bases: object

Base class for agent events.

type: str

timestamp: float

__init__(type: str, timestamp: float = <factory>) → None

class agenix.core.messages.AgentStartEvent(type: Literal['agent_start'] = 'agent_start', timestamp: float = <factory>)[source]

Bases: AgentEvent

Agent started processing.

type: Literal['agent_start'] = 'agent_start'

__init__(type: Literal['agent_start'] = 'agent_start', timestamp: float = <factory>) → None

class agenix.core.messages.AgentEndEvent(type: Literal['agent_end'] = 'agent_end', timestamp: float = <factory>, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage] = <factory>)[source]

Bases: AgentEvent

Agent finished processing.

type: Literal['agent_end'] = 'agent_end'

messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage]

__init__(type: Literal['agent_end'] = 'agent_end', timestamp: float = <factory>, messages: List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage] = <factory>) → None

class agenix.core.messages.TurnStartEvent(type: Literal['turn_start'] = 'turn_start', timestamp: float = <factory>)[source]

Bases: AgentEvent

Turn started.

type: Literal['turn_start'] = 'turn_start'

__init__(type: Literal['turn_start'] = 'turn_start', timestamp: float = <factory>) → None

class agenix.core.messages.TurnEndEvent(type: Literal['turn_end'] = 'turn_end', timestamp: float = <factory>, message: AssistantMessage | None = None, tool_results: List[ToolResultMessage] = <factory>)[source]

Bases: AgentEvent

Turn ended.

type: Literal['turn_end'] = 'turn_end'

message: AssistantMessage | None = None

tool_results: List[ToolResultMessage]

__init__(type: Literal['turn_end'] = 'turn_end', timestamp: float = <factory>, message: AssistantMessage | None = None, tool_results: List[ToolResultMessage] = <factory>) → None

class agenix.core.messages.MessageStartEvent(type: Literal['message_start'] = 'message_start', timestamp: float = <factory>, message: AssistantMessage | None = None)[source]

Bases: AgentEvent

Message streaming started.

type: Literal['message_start'] = 'message_start'

message: AssistantMessage | None = None

__init__(type: Literal['message_start'] = 'message_start', timestamp: float = <factory>, message: AssistantMessage | None = None) → None

class agenix.core.messages.MessageUpdateEvent(type: Literal['message_update'] = 'message_update', timestamp: float = <factory>, message: AssistantMessage | None = None, delta: str = '')[source]

Bases: AgentEvent

Message content updated.

type: Literal['message_update'] = 'message_update'

message: AssistantMessage | None = None

delta: str = ''

__init__(type: Literal['message_update'] = 'message_update', timestamp: float = <factory>, message: AssistantMessage | None = None, delta: str = '') → None

class agenix.core.messages.MessageEndEvent(type: Literal['message_end'] = 'message_end', timestamp: float = <factory>, message: AssistantMessage | None = None)[source]

Bases: AgentEvent

Message streaming ended.

type: Literal['message_end'] = 'message_end'

message: AssistantMessage | None = None

__init__(type: Literal['message_end'] = 'message_end', timestamp: float = <factory>, message: AssistantMessage | None = None) → None

class agenix.core.messages.ToolExecutionStartEvent(type: Literal['tool_execution_start'] = 'tool_execution_start', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', args: Dict[str, ~typing.Any]=<factory>)[source]

Bases: AgentEvent

Tool execution started.

type: Literal['tool_execution_start'] = 'tool_execution_start'

tool_call_id: str = ''

tool_name: str = ''

args: Dict[str, Any]

__init__(type: Literal['tool_execution_start'] = 'tool_execution_start', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', args: Dict[str, ~typing.Any]=<factory>) → None

class agenix.core.messages.ToolExecutionUpdateEvent(type: Literal['tool_execution_update'] = 'tool_execution_update', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', partial_result: Any = None)[source]

Bases: AgentEvent

Tool execution progress update.

type: Literal['tool_execution_update'] = 'tool_execution_update'

tool_call_id: str = ''

tool_name: str = ''

partial_result: Any = None

__init__(type: Literal['tool_execution_update'] = 'tool_execution_update', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', partial_result: Any = None) → None

class agenix.core.messages.ToolExecutionEndEvent(type: Literal['tool_execution_end'] = 'tool_execution_end', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', result: str | ToolResult | Any = None, is_error: bool = False)[source]

Bases: AgentEvent

Tool execution completed.

type: Literal['tool_execution_end'] = 'tool_execution_end'

tool_call_id: str = ''

tool_name: str = ''

result: str | ToolResult | Any = None

is_error: bool = False

__init__(type: Literal['tool_execution_end'] = 'tool_execution_end', timestamp: float = <factory>, tool_call_id: str = '', tool_name: str = '', result: str | ToolResult | Any = None, is_error: bool = False) → None

Session Management

Session management with persistence.

class agenix.core.session.SessionManager(session_dir: str = '.agenix')[source]

Bases: object

Manage persistent agent sessions.

__init__(session_dir: str = '.agenix')[source]

create_session(name: str | None = None) → str[source]

Create a new session.

Parameters:: name – Optional session name
Returns:: Session ID

save_message(session_id: str, message: UserMessage | AssistantMessage | ToolResultMessage | SystemMessage) → None[source]

Save a message to session.

Parameters:

session_id – Session ID
message – Message to save

load_session(session_id: str) → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]

Load session messages.

Parameters:: session_id – Session ID
Returns:: List of messages

list_sessions() → List[Dict[str, Any]][source]

List all sessions.

Returns:: List of session metadata

delete_session(session_id: str) → None[source]

Delete a session.

Parameters:: session_id – Session ID

Skills

Skill management system following the Anthropic Agent Skills standard.

This module implements the Agent Skills specification, providing discovery, loading, and management of skills that extend Claude’s capabilities.

Skills are defined using markdown files with YAML frontmatter containing: - name (required): Unique identifier (lowercase, hyphens, 1-64 chars) - description (required): Brief description of what the skill does - license (optional): License terms for the skill - compatibility (optional): Compatibility requirements - allowed-tools (optional): List of tools the skill can use - disable-model-invocation (optional): If true, skill is not shown to model - metadata (optional): Additional custom metadata

Examples

>>> manager = SkillManager(skill_dirs=['/path/to/skills'])
>>> skill = manager.get_skill('pdf')
>>> print(skill.description)
'PDF manipulation toolkit...'

class agenix.core.skills.Skill(name: str, description: str, file_path: str, content: str, license: str | None = None, compatibility: str | None = None, allowed_tools: List[str] | None = None, metadata: Dict[str, ~typing.Any]=<factory>, disable_model_invocation: bool = False)[source]

Bases: object

Represents a skill with its metadata and content.

A skill extends Claude’s capabilities by providing specialized knowledge or workflows. Skills follow the Anthropic Agent Skills standard.

name

Unique skill identifier (lowercase, hyphens, 1-64 chars)

Type:: str

description

Brief description of the skill’s purpose

Type:: str

file_path

Absolute path to the skill’s markdown file

Type:: str

content

Full markdown content including frontmatter

Type:: str

license

License terms for the skill

Type:: Optional[str]

compatibility

Compatibility requirements (e.g., “Python 3.8+”)

Type:: Optional[str]

allowed_tools

List of tools this skill can use

Type:: Optional[List[str]]

metadata

Additional custom metadata

Type:: Dict[str, Any]

disable_model_invocation

If True, skill is not shown to model

Type:: bool

Examples

>>> skill = Skill(
...     name="pdf",
...     description="PDF manipulation toolkit",
...     file_path="/skills/pdf/SKILL.md",
...     content="---\nname: pdf\n...",
...     allowed_tools=["Bash(pdf:*)"]
... )
>>> print(skill.name)
'pdf'

name: str

description: str

file_path: str

content: str

license: str | None = None

compatibility: str | None = None

allowed_tools: List[str] | None = None

metadata: Dict[str, Any]

disable_model_invocation: bool = False

__init__(name: str, description: str, file_path: str, content: str, license: str | None = None, compatibility: str | None = None, allowed_tools: List[str] | None = None, metadata: Dict[str, ~typing.Any]=<factory>, disable_model_invocation: bool = False) → None

class agenix.core.skills.SkillManager(skill_dirs: List[str] | None = None)[source]

Bases: object

Manages discovery, loading, and access to skills.

The SkillManager discovers skills from configured directories, parses their frontmatter, validates them according to the Agent Skills standard, and provides access to skill metadata and content.

Skill discovery looks for: - Direct .md files in the root of skill directories - SKILL.md files in subdirectories (subdirectory name should match skill name)

Default skill directories (if no custom dirs specified): - Global: ~/.agenix/skills/ - Project: .agenix/skills/

skill_dirs

List of directories to search for skills

Type:: List[str]

skills

Dictionary mapping skill names to Skill objects

Type:: Dict[str, Skill]

Examples

>>> # Use default directories
>>> manager = SkillManager()
>>> print(len(manager.list_skills()))
5

>>> # Custom directories
>>> manager = SkillManager(skill_dirs=['/my/skills', '/other/skills'])
>>> skill = manager.get_skill('pdf')
>>> content = manager.load_skill_content('pdf')

>>> # Generate summary for system prompt
>>> summary = manager.get_skills_summary()
>>> print(summary)
<available_skills>
  <skill name="pdf">
    PDF manipulation toolkit...
  </skill>
</available_skills>

__init__(skill_dirs: List[str] | None = None)[source]

Initialize skill manager and load skills.

Parameters:: skill_dirs – List of directories to search for skills. If None, uses default directories (~/.agenix/skills/ and .agenix/skills/)

Examples

>>> manager = SkillManager()  # Use defaults
>>> manager = SkillManager(skill_dirs=['/custom/path'])  # Custom

skills: Dict[str, Skill]

get_skill(name: str) → Skill | None[source]

Get a skill by name.

Parameters:: name – Skill name
Returns:: Skill object if found, None otherwise

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> skill = manager.get_skill('pdf')
>>> if skill:
...     print(skill.description)

list_skills() → List[Skill][source]

List all available skills.

Returns:: List of all loaded Skill objects

Examples

>>> manager = SkillManager()
>>> for skill in manager.list_skills():
...     print(f"{skill.name}: {skill.description}")

get_skills_summary() → str[source]

Get a summary of available skills for system prompt.

Generates an XML-formatted summary of skills that are not disabled. This summary is injected into the system prompt to make Claude aware of available skills.

Returns:: XML-formatted string with skill names and descriptions, or empty string if no visible skills

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> summary = manager.get_skills_summary()
>>> print(summary)
<available_skills>
  <skill name="pdf">
    PDF manipulation toolkit...
  </skill>
</available_skills>

Note

Skills with disable_model_invocation=True are excluded from the summary. These skills exist but are not shown to the model.

load_skill_content(name: str) → str | None[source]

Load full content of a skill.

Returns the complete markdown content including frontmatter. This is used for progressive disclosure - only loading full skill content when the model needs it.

Parameters:: name – Skill name
Returns:: Full skill markdown content, or None if skill not found

Examples

>>> manager = SkillManager(skill_dirs=['/skills'])
>>> content = manager.load_skill_content('pdf')
>>> if content:
...     print("Loaded skill documentation")

add_skill_dir(directory: str)[source]

Add a skill directory and load skills from it.

Dynamically adds a new skill directory to search and discovers any skills in that directory.

Parameters:: directory – Path to skill directory

Examples

>>> manager = SkillManager()
>>> manager.add_skill_dir('/custom/skills')
>>> print(len(manager.list_skills()))

SDK

Programmatic SDK for using agenix as a library.

Example usage:

```python import asyncio from agenix import create_session

async def main():

# Create a session session = await create_session(

api_key=”your-api-key”, model=”gpt-4o”, working_dir=”/path/to/project”

)

# Send a prompt response = await session.prompt(“What files are in the current directory?”) print(response)

# Continue conversation response = await session.prompt(“Can you read the README?”) print(response)

# Get conversation history messages = session.get_messages() print(f”Conversation has {len(messages)} messages”)

if __name__ == “__main__”:

asyncio.run(main())

```

async agenix.sdk.create_session(api_key: str | None = None, base_url: str | None = None, model: str = 'gpt-4o', system_prompt: str | None = None, working_dir: str = '.', max_turns: int = 10, enable_extensions: bool = True) → AgentSession[source]

Create an agent session for programmatic usage.

Parameters:

api_key – API key for the LLM provider (or set OPENAI_API_KEY env var)
base_url – Optional API base URL
model – Model to use (default: gpt-4o)
system_prompt – Optional custom system prompt
working_dir – Working directory for file operations
max_turns – Maximum conversation turns per prompt
enable_extensions – Whether to load extensions

Returns:

AgentSession instance

Example

>>> import asyncio
>>> from agenix import create_session
>>>
>>> async def main():
...     session = await create_session(
...         api_key="your-key",
...         model="gpt-4o"
...     )
...     response = await session.prompt("Hello!")
...     print(response)
...
>>> asyncio.run(main())

class agenix.sdk.AgentSession(agent: Agent, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

Bases: object

High-level agent session for programmatic usage.

Wraps the low-level Agent class with a simpler API.

__init__(agent: Agent, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

async prompt(message: str) → str[source]

Send a prompt to the agent and return the response.

Parameters:: message – User message/prompt
Returns:: Agent’s text response

get_messages() → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]

Get conversation history.

Returns:: List of messages in the conversation

clear_messages() → None[source]: Clear conversation history.

async close() → None[source]: Close the session and cleanup.

async agenix.sdk.create_aos_session(api_key: str | None = None, base_url: str | None = None, model: str = 'gpt-4o', system_prompt: str | None = None, working_dir: str = '.', max_turns: int = 10, enable_extensions: bool = True, node_id: str = 'local', agent_id: str = 'main') → AOSAgentSession[source]

Create an AOS agent session for programmatic usage.

Uses the Agent OS architecture with MessageBus, DirectChannel, and AOSAgentRuntime.

Parameters:

api_key – API key for the LLM provider (or set OPENAI_API_KEY env var)
base_url – Optional API base URL
model – Model to use (default: gpt-4o)
system_prompt – Optional custom system prompt
working_dir – Working directory for file operations
max_turns – Maximum conversation turns per prompt
enable_extensions – Whether to load extensions
node_id – Node ID for AOS addressing (default: “local”)
agent_id – Agent ID for AOS addressing (default: “main”)

Returns:

AOSAgentSession instance

Example

>>> import asyncio
>>> from agenix import create_aos_session
>>>
>>> async def main():
...     session = await create_aos_session(
...         api_key="your-key",
...         model="gpt-4o"
...     )
...     response = await session.prompt("Hello!")
...     print(response)
...     await session.close()
...
>>> asyncio.run(main())

class agenix.sdk.AOSAgentSession(bus: Any, channel: Any, runtime: Any, agent_address: Any, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

Bases: object

High-level AOS agent session for programmatic usage.

Uses the Agent OS architecture with MessageBus, DirectChannel, and AOSAgentRuntime. Provides the same API as AgentSession but with AOS benefits: - Better modularity and separation of concerns - Support for multiple channels and agents - Network-ready architecture - Enhanced observability

__init__(bus: Any, channel: Any, runtime: Any, agent_address: Any, tools: List[Any], extension_runner: ExtensionRunner | None = None, working_dir: str = '.')[source]

async prompt(message: str) → str[source]

Send a prompt to the agent and return the response.

Parameters:: message – User message/prompt
Returns:: Agent’s text response

get_messages() → List[UserMessage | AssistantMessage | ToolResultMessage | SystemMessage][source]

Get conversation history.

Returns:: List of messages in the conversation

clear_messages() → None[source]: Clear conversation history.

async close() → None[source]: Close the session and cleanup AOS components.

class agenix.sdk.ToolDefinition(name: str, description: str, parameters: Dict[str, Any], execute: Callable[[...], Awaitable[str]])[source]

Bases: object

Definition for a custom tool.

name: str

description: str

parameters: Dict[str, Any]

execute: Callable[[...], Awaitable[str]]

__init__(name: str, description: str, parameters: Dict[str, Any], execute: Callable[[...], Awaitable[str]]) → None

class agenix.sdk.CommandDefinition(name: str, description: str, handler: Callable[[ExtensionContext, str], Awaitable[None]])[source]

Bases: object

Definition for a custom command.

name: str

description: str

handler: Callable[[ExtensionContext, str], Awaitable[None]]

__init__(name: str, description: str, handler: Callable[[ExtensionContext, str], Awaitable[None]]) → None

class agenix.sdk.ExtensionAPI(*args, **kwargs)[source]

Bases: Protocol

API provided to extensions during setup.

Extensions use this to register tools, commands, and event handlers.

register_tool(tool: ToolDefinition) → None[source]: Register a custom tool that the LLM can call.

register_command(command: CommandDefinition) → None[source]: Register a custom command.

on(event_type: EventType, handler: Callable[[Event, ExtensionContext], Awaitable[None]]) → None[source]: Subscribe to an event.

notify(message: str, type: str = 'info') → None[source]: Show a notification.

__init__(*args, **kwargs)

class agenix.sdk.EventType(value)[source]

Bases: str, Enum

Agent lifecycle event types.

SESSION_START = 'session_start'

SESSION_END = 'session_end'

AGENT_START = 'agent_start'

AGENT_END = 'agent_end'

TURN_START = 'turn_start'

TURN_END = 'turn_end'

TOOL_CALL = 'tool_call'

TOOL_RESULT = 'tool_result'

USER_INPUT = 'user_input'

CLI

Main entry point for agenix.

agenix.cli.parse_args()[source]: Parse command line arguments.

agenix.cli.get_default_model() → str[source]: Get default model.

agenix.cli.get_default_system_prompt(tools: list) → str[source]

Get default system prompt with dynamic guidelines based on available tools.

Parameters:: tools – List of available tool instances

agenix.cli.validate_config(args, renderer: CLIRenderer = None) → tuple[source]

Validate configuration and return api_key, base_url, model.

Parameters:

args – Command-line arguments
renderer – CLI renderer for interactive input (optional)

Returns:

(api_key, base_url, model)

Return type:

tuple

Raises:

ValueError – If configuration is invalid and no renderer provided

agenix.cli.main()[source]: Main entry point.

async agenix.cli.process_single_message(agent, message: str, renderer: CLIRenderer)[source]: Process a single message in non-interactive mode.

Tools

Base Tool

Base tool interface.

class agenix.tools.base.ToolResult(content: str | List[TextContent | ImageContent], details: Dict[str, Any] | None = None, is_error: bool = False)[source]

Bases: object

Tool execution result.

content: str | List[TextContent | ImageContent]

details: Dict[str, Any] | None = None

is_error: bool = False

__init__(content: str | List[TextContent | ImageContent], details: Dict[str, Any] | None = None, is_error: bool = False) → None

class agenix.tools.base.Tool(name: str, description: str, parameters: Dict[str, Any])[source]

Bases: ABC

Abstract tool interface.

__init__(name: str, description: str, parameters: Dict[str, Any])[source]

abstractmethod async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]

Execute the tool with given arguments.

Parameters:

tool_call_id – Unique ID for this tool call
arguments – Tool arguments (must be a dict)
on_update – Optional callback for progress updates

Returns:

ToolResult with content and optional details

to_dict() → Dict[str, Any][source]: Convert tool to API format.

validate_arguments(arguments: Dict[str, Any]) → None[source]: Validate tool arguments against schema.

Read Tool

Read file tool.

class agenix.tools.read.ReadTool(working_dir: str = '.')[source]

Bases: Tool

Read file content with image support.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute read operation.

Write Tool

Write file tool.

class agenix.tools.write.WriteTool(working_dir: str = '.')[source]

Bases: Tool

Write content to a file.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute write operation.

Edit Tool

Edit file tool with exact string replacement.

class agenix.tools.edit.EditTool(working_dir: str = '.')[source]

Bases: Tool

Edit file by replacing exact strings.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute edit operation.

Bash Tool

Bash command execution tool.

class agenix.tools.bash.BashTool(working_dir: str = '.', timeout: int = 120)[source]

Bases: Tool

Execute bash commands.

__init__(working_dir: str = '.', timeout: int = 120)[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute bash command.

Grep Tool

Grep/search tool.

class agenix.tools.grep.GrepTool(working_dir: str = '.')[source]

Bases: Tool

Search for patterns in files.

__init__(working_dir: str = '.')[source]

async execute(tool_call_id: str, arguments: Dict[str, Any], on_update: Callable[[str], None] | None = None) → ToolResult[source]: Execute grep operation.

Extensions System

Extension Types

Extension system types and interfaces.

Extensions are Python modules that can: - Subscribe to agent lifecycle events - Register custom LLM-callable tools - Register custom commands - Access agent context and UI primitives

class agenix.extensions.types.EventType(value)[source]

Bases: str, Enum

Agent lifecycle event types.

SESSION_START = 'session_start'

SESSION_END = 'session_end'

AGENT_START = 'agent_start'

AGENT_END = 'agent_end'

TURN_START = 'turn_start'

TURN_END = 'turn_end'

TOOL_CALL = 'tool_call'

TOOL_RESULT = 'tool_result'

USER_INPUT = 'user_input'

class agenix.extensions.types.Event(type: EventType, data: Dict[str, Any])[source]

Bases: object

Base event class.

type: EventType

data: Dict[str, Any]

__init__(type: EventType, data: Dict[str, Any]) → None

class agenix.extensions.types.SessionStartEvent[source]

Bases: Event

Fired when session starts.

__init__()[source]

class agenix.extensions.types.SessionEndEvent[source]

Bases: Event

Fired when session ends.

__init__()[source]

class agenix.extensions.types.AgentStartEvent(prompt: str)[source]

Bases: Event

Fired when agent loop starts.

__init__(prompt: str)[source]

property prompt: str

class agenix.extensions.types.AgentEndEvent(messages: List[Any])[source]

Bases: Event

Fired when agent loop ends.

__init__(messages: List[Any])[source]

property messages: List[Any]

class agenix.extensions.types.TurnStartEvent(turn_index: int)[source]

Bases: Event

Fired at the start of each turn.

__init__(turn_index: int)[source]

property turn_index: int

class agenix.extensions.types.TurnEndEvent(turn_index: int, message: Any)[source]

Bases: Event

Fired at the end of each turn.

__init__(turn_index: int, message: Any)[source]

property turn_index: int

property message: Any

class agenix.extensions.types.ToolCallEvent(tool_name: str, args: Dict[str, Any])[source]

Bases: Event

Fired before a tool executes.

__init__(tool_name: str, args: Dict[str, Any])[source]

property tool_name: str

property args: Dict[str, Any]

class agenix.extensions.types.ToolResultEvent(tool_name: str, result: Any, is_error: bool)[source]

Bases: Event

Fired after a tool executes.

__init__(tool_name: str, result: Any, is_error: bool)[source]

property tool_name: str

property result: Any

property is_error: bool

class agenix.extensions.types.UserInputEvent(text: str)[source]

Bases: Event

Fired when user provides input.

__init__(text: str)[source]

property text: str

class agenix.extensions.types.ExtensionContext(agent: Agent, cwd: str, tools: List[Tool])[source]

Bases: object

Context passed to extension event handlers.

Provides access to agent state and operations.

__init__(agent: Agent, cwd: str, tools: List[Tool])[source]

property messages: List[Any]: Get current conversation messages.

notify(message: str, type: str = 'info') → None[source]: Show a notification to the user.

class agenix.extensions.types.ToolDefinition(name: str, description: str, parameters: Dict[str, Any], execute: Callable[[...], Awaitable[str]])[source]

Bases: object

Definition for a custom tool.

name: str

description: str

parameters: Dict[str, Any]

execute: Callable[[...], Awaitable[str]]

__init__(name: str, description: str, parameters: Dict[str, Any], execute: Callable[[...], Awaitable[str]]) → None

class agenix.extensions.types.CommandDefinition(name: str, description: str, handler: Callable[[ExtensionContext, str], Awaitable[None]])[source]

Bases: object

Definition for a custom command.

name: str

description: str

handler: Callable[[ExtensionContext, str], Awaitable[None]]

__init__(name: str, description: str, handler: Callable[[ExtensionContext, str], Awaitable[None]]) → None

class agenix.extensions.types.ExtensionAPI(*args, **kwargs)[source]

Bases: Protocol

API provided to extensions during setup.

Extensions use this to register tools, commands, and event handlers.

register_tool(tool: ToolDefinition) → None[source]: Register a custom tool that the LLM can call.

register_command(command: CommandDefinition) → None[source]: Register a custom command.

on(event_type: EventType, handler: Callable[[Event, ExtensionContext], Awaitable[None]]) → None[source]: Subscribe to an event.

notify(message: str, type: str = 'info') → None[source]: Show a notification.

__init__(*args, **kwargs)

class agenix.extensions.types.LoadedExtension(path: str, name: str, tools: Dict[str, ToolDefinition], commands: Dict[str, CommandDefinition], handlers: Dict[EventType, List[Callable[[Event, ExtensionContext], Awaitable[None]]]])[source]

Bases: object

A loaded extension with its registered items.

path: str

name: str

tools: Dict[str, ToolDefinition]

commands: Dict[str, CommandDefinition]

handlers: Dict[EventType, List[Callable[[Event, ExtensionContext], Awaitable[None]]]]

__init__(path: str, name: str, tools: Dict[str, ToolDefinition], commands: Dict[str, CommandDefinition], handlers: Dict[EventType, List[Callable[[Event, ExtensionContext], Awaitable[None]]]]) → None

Extension Loader

Extension loader - discovers and loads Python extension modules.

class agenix.extensions.loader.ExtensionLoaderAPI(extension: LoadedExtension)[source]

Bases: object

Implementation of ExtensionAPI for use during extension loading.

__init__(extension: LoadedExtension)[source]

register_tool(tool: ToolDefinition) → None[source]: Register a custom tool.

register_command(command: CommandDefinition) → None[source]: Register a custom command.

on(event_type: EventType, handler: Callable[[Event, ExtensionContext], Awaitable[None]] | None = None)[source]

Subscribe to an event.

Can be used as a decorator or direct call:

# As decorator: @agenix.on(EventType.SESSION_START) async def handler(event, ctx):

pass

# Direct call: agenix.on(EventType.SESSION_START, handler)

notify(message: str, type: str = 'info') → None[source]: Show a notification.

agenix.extensions.loader.discover_extensions(directory: str) → List[str][source]

Discover extension files in a directory.

Returns list of absolute paths to .py files.

agenix.extensions.loader.load_extension_module(file_path: str) → Callable[[ExtensionAPI], Awaitable[None]] | None[source]

Load a Python extension module and return its setup function.

Returns:: The setup() function from the module, or None if not found.

async agenix.extensions.loader.load_extension(file_path: str) → LoadedExtension | None[source]

Load an extension from a file path.

Returns:: LoadedExtension instance, or None if loading failed.

async agenix.extensions.loader.discover_and_load_extensions(cwd: str, agenix_dir: str | None = None) → List[LoadedExtension][source]

Discover and load extensions from standard locations.

Loads from (in order): 1. Global: ~/.agenix/extensions/ 2. Project: .agenix/extensions/

Parameters:

cwd – Current working directory (for project-local extensions)
agenix_dir – Global agenix directory (default: ~/.agenix)

Returns:

List of successfully loaded extensions.

Extension Runner

Extension runner - executes extensions and manages their lifecycle.

class agenix.extensions.runner.ExtensionRunner(extensions: List[LoadedExtension], context: ExtensionContext)[source]

Bases: object

Manages execution of loaded extensions.

__init__(extensions: List[LoadedExtension], context: ExtensionContext)[source]

get_tools() → Dict[str, ToolDefinition][source]: Get all registered custom tools from extensions.

get_commands() → Dict[str, CommandDefinition][source]: Get all registered commands from extensions.

async emit(event: Event) → None[source]

Emit an event to all registered handlers.

Parameters:: event – The event to emit.

async execute_command(command_name: str, args: str) → bool[source]

Execute a registered extension command.

Parameters:

command_name – Name of the command
args – Command arguments as string

Returns:

True if command was found and executed, False otherwise.

has_handlers(event_type: EventType) → bool[source]: Check if any extension has handlers for an event type.

get_extension_paths() → List[str][source]: Get list of loaded extension paths.

get_extension_names() → List[str][source]: Get list of loaded extension names.

UI Components

CLI Interface

CLI interface for agenix.

class agenix.ui.cli.CLIRenderer[source]

Bases: object

Render agent events to terminal.

__init__()[source]

render_event(event: AgentStartEvent | AgentEndEvent | TurnStartEvent | TurnEndEvent | MessageStartEvent | MessageUpdateEvent | MessageEndEvent | ToolExecutionStartEvent | ToolExecutionUpdateEvent | ToolExecutionEndEvent) → None[source]: Render an event to the console.

render_message(role: str, content: str, is_error: bool = False) → None[source]: Render a complete message in a box.

render_error(error: str) → None[source]: Render an error.

render_welcome(model: str = None, tools=None, skills=None) → None[source]: Render welcome banner with ASCII art.

prompt_config_input(field_name: str, description: str, is_secret: bool = False) → str[source]

Prompt user for configuration input.

Parameters:

field_name – Name of the configuration field
description – Description of what this field is for
is_secret – If True, input will be hidden

Returns:

The user input

prompt(text: str = '>') → str[source]: Show input prompt with better unicode support.

clear() → None[source]: Clear the screen.

class agenix.ui.cli.CLI(renderer: CLIRenderer | None = None)[source]

Bases: object

Main CLI interface.

__init__(renderer: CLIRenderer | None = None)[source]

run_interactive(agent, tools=None, model=None, skills=None, show_welcome=True) → None[source]: Run interactive chat loop.

async process_message(agent, user_input: str) → None[source]: Process a user message.

handle_command(command: str, agent) → bool[source]

Handle CLI commands.

Returns:: True to continue, False to exit

list_sessions() → None[source]: List saved sessions.

load_session(agent, session_id: str) → None[source]: Load a session.