BerriAI/agentchat

agentchat

Quickstart

1. Install

git clone https://github.com/BerriAI/agentchat.git
cd agentchat
python -m pip install -e ".[openai]"

2. Usage

import asyncio
import os

from agents import Agent, Runner, SQLiteSession # import `Agent` from OpenAI Agents SDK

from agentchat import AgentChat, MessageContext
from agentchat.channels.slack import Slack

app = AgentChat(channels=[Slack.from_env(ack_emoji=None)])

agent = Agent(name="Assistant", model=os.environ["OPENAI_MODEL"])
sessions = {}


@app.on_message
async def respond(context: MessageContext) -> str:
    key = context.conversation.id
    if key not in sessions:
        sessions[key] = SQLiteSession(key)

    result = await Runner.run(
        agent,
        input=context.message.text,
        session=sessions[key],
    )
    return str(result.final_output)


async def main():
    try:
        await app.run()
    finally:
        await app.close()


asyncio.run(main())

Other agent SDKs

1. Claude Agent SDK

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query

sessions = {}


@app.on_message
async def respond(context: MessageContext) -> str:
    key = context.conversation.id
    options = ClaudeAgentOptions(resume=sessions.get(key), tools=[])

    result = None
    async for event in query(prompt=context.message.text, options=options):
        if isinstance(event, ResultMessage):
            sessions[key] = event.session_id
            result = event

    if result is None or result.is_error or not result.result:
        raise RuntimeError("Claude did not return a successful answer")
    return result.result

2. Langchain Deep Agents

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model=os.environ["DEEP_AGENT_MODEL"],
    checkpointer=InMemorySaver(),
)


@app.on_message
async def respond(context: MessageContext) -> str:
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": context.message.text}]},
        config={"configurable": {"thread_id": context.conversation.id}},
    )
    if result.get("__interrupt__"):
        raise RuntimeError("This example does not handle approval requests")
    answer = result["messages"][-1]
    if answer.type != "ai" or answer.tool_calls or not answer.text:
        raise RuntimeError("The agent did not return a final text answer")
    return answer.text

3. Pydantic AI

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage

agent = Agent(f"openai:{os.environ['OPENAI_MODEL']}", output_type=str)
histories: dict[str, list[ModelMessage]] = {}


@app.on_message
async def respond(context: MessageContext) -> str:
    key = context.conversation.id
    result = await agent.run(
        context.message.text,
        message_history=histories.get(key, []),
    )
    histories[key] = result.all_messages()
    return result.output

4. LangGraph SDK

from langgraph_sdk import get_client

client = get_client(
    url=os.environ["LANGGRAPH_URL"],
    api_key=os.getenv("LANGSMITH_API_KEY"),
)
threads: dict[str, str] = {}


@app.on_message
async def respond(context: MessageContext) -> str:
    key = context.conversation.id
    if key not in threads:
        thread = await client.threads.create()
        threads[key] = thread["thread_id"]

    result = await client.runs.wait(
        threads[key],
        os.environ["LANGGRAPH_ASSISTANT_ID"],
        input={"messages": [{"role": "user", "content": context.message.text}]},
    )
    if result.get("__interrupt__"):
        raise RuntimeError("The graph is waiting for approval")
    answer = result["messages"][-1]
    if answer["type"] != "ai" or answer.get("tool_calls"):
        raise RuntimeError("The graph did not return a final answer")
    content = answer["content"]
    if not isinstance(content, str) or not content:
        raise RuntimeError("This example expects a text answer")
    return content

Supported channels