pip install ag_kit_py

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
)

from ag_kit_py.server import AGKitAPIApp

def create_agent():
    return {"agent": agent}

AGKitAPIApp().run(create_agent, port=8000)

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

@property
def name(self) -> str:
    """Get the agent name."""
    return self._name

@property
def description(self) -> str:
    """Get the agent description."""
    return self._description

@property
def agent(self) -> Any:
    """Get the underlying agent instance."""
    return self._agent

@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
    """

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

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

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

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()

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

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)}
        )

def destroy(self) -> None:
    """Always implement cleanup."""
    if hasattr(self._agent, 'close'):
        self._agent.close()
    if hasattr(self, '_db_connection'):
        self._db_connection.close()

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

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, ...)