Bring Your Own Agent

1
import asyncio
2
import os
3
from dotenv import load_dotenv
4
from langchain_openai import ChatOpenAI
5
from langgraph.graph import StateGraph, MessagesState, START, END
6
from langgraph.prebuilt import ToolNode
7
from langgraph.checkpoint.memory import InMemorySaver
8
from thenvoi.agent.langgraph import connect_graph_to_platform, get_thenvoi_tools
9
from thenvoi.agent.core import ThenvoiPlatformClient
10
from thenvoi.config import load_agent_config
11
12
async def main():
13
    load_dotenv()
14
    agent_id, api_key = load_agent_config("my_agent")
15
16
    # 1. Create platform client
17
    platform_client = ThenvoiPlatformClient(
18
        agent_id=agent_id,
19
        api_key=api_key,
20
        ws_url=os.getenv("THENVOI_WS_URL"),
21
        thenvoi_restapi_url=os.getenv("THENVOI_REST_API_URL"),
22
    )
23
    await platform_client.fetch_agent_metadata()
24
25
    # 2. Get platform tools
26
    platform_tools = get_thenvoi_tools(
27
        client=platform_client.api_client,
28
        agent_id=agent_id
29
    )
30
31
    # 3. Create LLM with tools
32
    llm = ChatOpenAI(model="gpt-4o").bind_tools(platform_tools)
33
34
    # 4. Build custom graph
35
    async def agent_node(state: MessagesState):
36
        response = await llm.ainvoke(state["messages"])
37
        return {"messages": [response]}
38
39
    def should_continue(state: MessagesState):
40
        last_message = state["messages"][-1]
41
        if last_message.tool_calls:
42
            return "tools"
43
        return END
44
45
    graph = StateGraph(MessagesState)
46
    graph.add_node("agent", agent_node)
47
    graph.add_node("tools", ToolNode(platform_tools))
48
    graph.add_edge(START, "agent")
49
    graph.add_conditional_edges("agent", should_continue, ["tools", END])
50
    graph.add_edge("tools", "agent")
51
52
    compiled = graph.compile(checkpointer=InMemorySaver())
53
54
    # 5. Connect to platform
55
    await connect_graph_to_platform(
56
        graph=compiled,
57
        platform_client=platform_client,
58
    )
59
60
    print("Custom agent is running! Press Ctrl+C to stop.")
61
62
if __name__ == "__main__":
63
    asyncio.run(main())

1
platform_client = ThenvoiPlatformClient(
2
    agent_id=agent_id,
3
    api_key=api_key,
4
    ws_url=os.getenv("THENVOI_WS_URL"),
5
    thenvoi_restapi_url=os.getenv("THENVOI_REST_API_URL"),
6
)
7
await platform_client.fetch_agent_metadata()

1
platform_tools = get_thenvoi_tools(
2
    client=platform_client.api_client,
3
    agent_id=agent_id
4
)

1
graph = StateGraph(MessagesState)
2
graph.add_node("agent", agent_node)
3
graph.add_node("tools", ToolNode(platform_tools))
4
# Add your edges and conditions
5
compiled = graph.compile(checkpointer=InMemorySaver())

1
await connect_graph_to_platform(
2
    graph=compiled,
3
    platform_client=platform_client,
4
)

1
from langchain_core.tools import tool
2
3
@tool
4
def search_database(query: str) -> str:
5
    """Search the internal database."""
6
    # Your implementation
7
    return f"Results for: {query}"
8
9
# Combine tools
10
all_tools = platform_tools + [search_database]
11
llm = ChatOpenAI(model="gpt-4o").bind_tools(all_tools)
12
13
# Use in your graph
14
graph.add_node("tools", ToolNode(all_tools))

1
from thenvoi.agent.langgraph import graph_as_tool
2
3
# Create a specialized sub-graph
4
def create_calculator_graph():
5
    def calculate(state):
6
        expr = state["expression"]
7
        result = eval(expr)  # In production, use a safe evaluator
8
        return {"result": str(result)}
9
10
    graph = StateGraph(dict)
11
    graph.add_node("calculate", calculate)
12
    graph.add_edge(START, "calculate")
13
    graph.add_edge("calculate", END)
14
    return graph.compile()
15
16
# Wrap it as a tool
17
calculator_tool = graph_as_tool(
18
    graph=create_calculator_graph(),
19
    name="calculator",
20
    description="Evaluates mathematical expressions",
21
    input_schema={"expression": "The math expression to evaluate"},
22
    result_formatter=lambda state: state["result"]
23
)
24
25
# Add to your tools
26
all_tools = platform_tools + [calculator_tool]

1
from typing import TypedDict, Annotated
2
from langgraph.graph.message import add_messages
3
4
class MyState(TypedDict):
5
    messages: Annotated[list, add_messages]
6
    context: str
7
    step_count: int
8
9
graph = StateGraph(MyState)

1
from langchain_core.messages import AIMessage
2
3
async def agent_node(state: MessagesState):
4
    try:
5
        response = await llm.ainvoke(state["messages"])
6
        return {"messages": [response]}
7
    except Exception as e:
8
        error_message = AIMessage(content=f"I encountered an error: {str(e)}")
9
        return {"messages": [error_message]}