Documentation Index
Fetch the complete documentation index at: https://docs.ag-kit.dev/llms.txt
Use this file to discover all available pages before exploring further.
ag_kit_py.agents
The Agents module provides the core abstractions and implementations for building AI agents with AG-Kit. It defines the base agent interface and provides framework-specific implementations like LangGraph.
Installation
Available Agents
| Agent Type | Class | Status | Description |
|---|
| LangGraph | LangGraphAgent | ✅ Available | LangGraph-based agent with stability patches |
| Base | BaseAgent | ✅ Available | Abstract base class for custom agents |
Quick Start
Creating a LangGraph Agent
from ag_kit_py.agents import LangGraphAgent
from ag_kit_py.providers import create_provider
from langgraph.graph import StateGraph, MessagesState
# Create provider
provider = create_provider("openai")
model = provider.get_langchain_model()
# Define chat node
def chat_node(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": [response]}
# Build workflow
workflow = StateGraph(MessagesState)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.set_finish_point("chat")
# Compile and create agent
graph = workflow.compile()
agent = LangGraphAgent(
name="ChatAgent",
description="A conversational agent",
graph=graph
)
Using with Server
from ag_kit_py.server import AgentServiceApp
def create_agent():
return {"agent": agent}
AgentServiceApp().run(create_agent, port=8000)
BaseAgent
Abstract base class that all agents must extend.
Class Definition
from abc import ABC, abstractmethod
from typing import Any, AsyncGenerator
from ag_ui.core import BaseEvent, RunAgentInput
class BaseAgent(ABC):
"""Abstract base class for all AG-Kit agents."""
def __init__(self, name: str, description: str, agent: Any):
"""Initialize the base agent.
Args:
name: Human-readable name for the agent
description: Detailed description of the agent's purpose
agent: The underlying agent implementation
"""
self._name = name
self._description = description
self._agent = agent
Properties
name
Get the agent name.
@property
def name(self) -> str:
"""Get the agent name."""
return self._name
description
Get the agent description.
@property
def description(self) -> str:
"""Get the agent description."""
return self._description
agent
Get the underlying agent implementation.
@property
def agent(self) -> Any:
"""Get the underlying agent instance."""
return self._agent
Abstract Methods
run()
Execute the agent with the given input.
@abstractmethod
async def run(
self,
run_input: RunAgentInput
) -> AsyncGenerator[BaseEvent, None]:
"""Execute the agent with the given input.
Args:
run_input: Input data for agent execution
Yields:
BaseEvent: Events representing execution progress
"""
Concrete Methods
destroy()
Clean up resources used by the agent.
def destroy(self) -> None:
"""Clean up resources used by the agent."""
pass
from ag_ui.core import RunAgentInput
from typing import Any, Dict, List
class RunAgentInput:
"""Input data for agent execution."""
messages: List[Dict[str, Any]] # Conversation messages
run_id: str # Unique run identifier
thread_id: str # Thread/conversation identifier
state: Dict[str, Any] # Agent state
context: List[Any] # Additional context
tools: List[Any] # Available tools
forwarded_props: Dict[str, Any] # Forwarded properties
BaseEvent
Base event class for agent execution events.
from ag_ui.core import BaseEvent, EventType
class BaseEvent:
"""Base event for agent execution."""
type: EventType # Event type
thread_id: str # Thread identifier
run_id: str # Run identifier
raw_event: Any # Raw event data
Event Types
from ag_ui.core import EventType
# Available event types
EventType.RUN_STARTED # Agent execution started
EventType.RUN_FINISHED # Agent execution finished
EventType.RUN_ERROR # Agent execution error
EventType.TEXT_MESSAGE_START # Text message started
EventType.TEXT_MESSAGE_CONTENT # Text message content chunk
EventType.TEXT_MESSAGE_END # Text message ended
EventType.TOOL_CALL_START # Tool call started
EventType.TOOL_CALL_ARGS # Tool call arguments (streaming)
EventType.TOOL_CALL_END # Tool call ended
EventType.TOOL_CALL_RESULT # Tool call result
EventType.CUSTOM # Custom event
Creating Custom Agents
To create a custom agent, extend BaseAgent and implement the run method:
from ag_kit_py.agents import BaseAgent
from ag_ui.core import RunAgentInput, BaseEvent, EventType
from typing import AsyncGenerator
class CustomAgent(BaseAgent):
"""Custom agent implementation."""
def __init__(self, name: str, description: str, custom_config: dict):
# Initialize your custom agent
agent_impl = create_custom_agent(custom_config)
super().__init__(name, description, agent_impl)
async def run(
self,
run_input: RunAgentInput
) -> AsyncGenerator[BaseEvent, None]:
"""Execute the custom agent."""
# Emit run started event
yield BaseEvent(
type=EventType.RUN_STARTED,
thread_id=run_input.thread_id,
run_id=run_input.run_id
)
try:
# Execute your agent logic
result = await self._agent.execute(run_input)
# Emit result events
yield BaseEvent(
type=EventType.TEXT_MESSAGE_CONTENT,
thread_id=run_input.thread_id,
run_id=run_input.run_id,
raw_event={"data": {"chunk": {"content": result}}}
)
# Emit run finished event
yield BaseEvent(
type=EventType.RUN_FINISHED,
thread_id=run_input.thread_id,
run_id=run_input.run_id
)
except Exception as e:
# Emit error event
yield BaseEvent(
type=EventType.RUN_ERROR,
thread_id=run_input.thread_id,
run_id=run_input.run_id,
raw_event={"error": str(e)}
)
def destroy(self) -> None:
"""Clean up resources."""
# Implement cleanup logic
self._agent.cleanup()
Best Practices
1. Use Type Hints
from ag_kit_py.agents import BaseAgent
from ag_ui.core import RunAgentInput, BaseEvent
from typing import AsyncGenerator
class MyAgent(BaseAgent):
async def run(
self,
run_input: RunAgentInput
) -> AsyncGenerator[BaseEvent, None]:
# Type hints improve code quality
pass
2. Handle Errors Gracefully
async def run(self, run_input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
try:
# Agent logic
yield event
except Exception as e:
# Always emit error events
yield BaseEvent(
type=EventType.RUN_ERROR,
thread_id=run_input.thread_id,
run_id=run_input.run_id,
raw_event={"error": str(e)}
)
3. Clean Up Resources
def destroy(self) -> None:
"""Always implement cleanup."""
if hasattr(self._agent, 'close'):
self._agent.close()
if hasattr(self, '_db_connection'):
self._db_connection.close()
4. Use Async/Await
async def run(self, run_input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
# Use async operations for better performance
result = await self._agent.async_execute(run_input)
yield result
5. Emit Appropriate Events
async def run(self, run_input: RunAgentInput) -> AsyncGenerator[BaseEvent, None]:
# Always emit RUN_STARTED
yield BaseEvent(type=EventType.RUN_STARTED, ...)
try:
# Emit progress events
yield BaseEvent(type=EventType.TEXT_MESSAGE_CONTENT, ...)
# Always emit RUN_FINISHED on success
yield BaseEvent(type=EventType.RUN_FINISHED, ...)
except Exception as e:
# Always emit RUN_ERROR on failure
yield BaseEvent(type=EventType.RUN_ERROR, ...)
