Creating Framework Integrations

1
async def on_message(self, msg, tools, history, ...):
2
    # Convert platform tools to LangChain format
3
    langchain_tools = agent_tools_to_langchain(tools)
4
5
    # Create graph with tools - graph handles tool execution internally
6
    graph = create_react_agent(llm, langchain_tools, checkpointer)
7
8
    # Stream events - LLM decides when to call thenvoi_send_message
9
    async for event in graph.astream_events({"messages": messages}, ...):
10
        await self._handle_stream_event(event, room_id, tools)

1
async def on_message(self, msg, tools, history, ...):
2
    # Get tool schemas in Anthropic format (sync method)
3
    tool_schemas = tools.get_tool_schemas("anthropic")
4
5
    # Tool execution loop
6
    while True:
7
        # Call LLM with tools
8
        response = await self.client.messages.create(
9
            model=self.model,
10
            messages=messages,
11
            tools=tool_schemas,
12
        )
13
14
        # Check if LLM wants to use tools
15
        if response.stop_reason != "tool_use":
16
            break  # LLM is done
17
18
        # IMPORTANT: Append assistant response (with tool_use blocks) to history
19
        messages.append({
20
            "role": "assistant",
21
            "content": response.content,  # Contains ToolUseBlock(s)
22
        })
23
24
        # Execute tool calls and collect results
25
        tool_results = []
26
        for block in response.content:
27
            if isinstance(block, ToolUseBlock):
28
                result = await tools.execute_tool_call(
29
                    block.name,  # e.g., "thenvoi_send_message"
30
                    block.input  # e.g., {"content": "Hello!", "mentions": ["User"]}
31
                )
32
                tool_results.append({
33
                    "type": "tool_result",
34
                    "tool_use_id": block.id,
35
                    "content": str(result),
36
                })
37
38
        # IMPORTANT: Append tool results to history
39
        messages.append({
40
            "role": "user",
41
            "content": tool_results,
42
        })

1
async def _handle_stream_event(
2
    self,
3
    event: dict,
4
    room_id: str,
5
    tools: AgentToolsProtocol,
6
) -> None:
7
    """Handle streaming events from framework."""
8
    event_type = event.get("event")
9
10
    if event_type == "on_tool_start":
11
        tool_name = event.get("name", "unknown")
12
        await tools.send_event(
13
            content=json.dumps(event, default=str),
14
            message_type="tool_call",
15
        )
16
17
    elif event_type == "on_tool_end":
18
        tool_name = event.get("name", "unknown")
19
        await tools.send_event(
20
            content=json.dumps(event, default=str),
21
            message_type="tool_result",
22
        )

1
async def _process_tool_calls(
2
    self,
3
    response: Message,
4
    tools: AgentToolsProtocol,
5
) -> list[dict]:
6
    """Execute tool calls and report events."""
7
    results = []
8
9
    for block in response.content:
10
        if not isinstance(block, ToolUseBlock):
11
            continue
12
13
        # Report tool call
14
        await tools.send_event(
15
            content=f"Calling {block.name}",
16
            message_type="tool_call",
17
            metadata={"tool": block.name, "input": block.input},
18
        )
19
20
        # Execute tool
21
        try:
22
            result = await tools.execute_tool_call(block.name, block.input)
23
            is_error = False
24
        except Exception as e:
25
            result = f"Error: {e}"
26
            is_error = True
27
28
        # Report result
29
        await tools.send_event(
30
            content=f"Result: {result}",
31
            message_type="tool_result",
32
            metadata={"tool": block.name, "is_error": is_error},
33
        )
34
35
        results.append({"tool_use_id": block.id, "content": str(result)})
36
37
    return results

1
async def on_message(self, msg, tools, ...):
2
    try:
3
        response = await self._call_llm(messages, tool_schemas)
4
        # ... process response ...
5
    except Exception as e:
6
        # Report error to platform
7
        await tools.send_event(
8
            content=f"Error: {e}",
9
            message_type="error",
10
        )
11
        raise  # Re-raise so message is marked as failed

1
import json
2
from typing import Any
3
4
MAX_TOOL_ITERS = 10
5
6
7
class ToolRecoverableError(Exception):
8
    """Errors the LLM can reasonably react to (bad args, not found, permission, etc)."""
9
10
11
class ToolInfraError(Exception):
12
    """Errors that indicate runtime/infrastructure problems (timeouts, 5xx, auth, etc)."""
13
14
15
def _maybe_json_loads(x: Any) -> Any:
16
    """Parse JSON string if needed; some LLM APIs return arguments as strings."""
17
    if isinstance(x, str):
18
        try:
19
            return json.loads(x)
20
        except json.JSONDecodeError:
21
            return x
22
    return x
23
24
25
def _tool_result_message(*, tool_call_id: str, content: str, is_error: bool) -> dict[str, Any]:
26
    """Build OpenAI-style tool result message."""
27
    prefix = "ERROR: " if is_error else ""
28
    return {
29
        "role": "tool",
30
        "tool_call_id": tool_call_id,
31
        "content": f"{prefix}{content}",
32
    }
33
34
35
class ManualToolLoopAdapter:
36
    def __init__(self, client, model: str):
37
        self.client = client
38
        self.model = model
39
40
    async def on_message(
41
        self, msg, tools, history, participants_msg, *, is_session_bootstrap: bool, room_id: str
42
    ) -> None:
43
        messages: list[dict[str, Any]] = list(history)
44
        messages.append({"role": "user", "content": msg.format_for_llm()})
45
46
        tool_schemas = tools.get_tool_schemas("openai")
47
48
        for i in range(MAX_TOOL_ITERS):
49
            # 1) Call the LLM
50
            try:
51
                resp = await self.client.responses.create(
52
                    model=self.model,
53
                    input=messages,
54
                    tools=tool_schemas,
55
                )
56
            except Exception as e:
57
                await tools.send_event(content=f"LLM call failed: {e}", message_type="error")
58
                raise
59
60
            # 2) Extract assistant content + tool calls
61
            assistant_content = getattr(resp, "output_text", None) or ""
62
            tool_calls = getattr(resp, "tool_calls", None) or []
63
64
            # 3) No tool calls = done
65
            if not tool_calls:
66
                if assistant_content:
67
                    messages.append({"role": "assistant", "content": assistant_content})
68
                return
69
70
            # 4) Append assistant message with tool calls
71
            messages.append({
72
                "role": "assistant",
73
                "content": assistant_content,
74
                "tool_calls": tool_calls,
75
            })
76
77
            # 5) Execute tool calls
78
            for tc in tool_calls:
79
                name = tc["name"]
80
                tool_call_id = tc["id"]
81
                args = _maybe_json_loads(tc.get("arguments", {}))
82
83
                await tools.send_event(
84
                    content=f"Calling {name}",
85
                    message_type="tool_call",
86
                    metadata={"tool": name, "input": args},
87
                )
88
89
                try:
90
                    result = await tools.execute_tool_call(name, args)
91
92
                    await tools.send_event(
93
                        content=f"{name} OK",
94
                        message_type="tool_result",
95
                        metadata={"tool": name, "is_error": False},
96
                    )
97
98
                    messages.append(_tool_result_message(
99
                        tool_call_id=tool_call_id,
100
                        content=str(result),
101
                        is_error=False,
102
                    ))
103
104
                except ToolRecoverableError as e:
105
                    # Recoverable: give LLM the error so it can decide what to do
106
                    await tools.send_event(
107
                        content=f"{name} recoverable error: {e}",
108
                        message_type="tool_result",
109
                        metadata={"tool": name, "is_error": True, "class": "recoverable"},
110
                    )
111
112
                    messages.append(_tool_result_message(
113
                        tool_call_id=tool_call_id,
114
                        content=str(e),
115
                        is_error=True,
116
                    ))
117
                    # Do NOT raise — let the loop continue so LLM can react
118
119
                except Exception as e:
120
                    # Infra failure: append result for context, then fail
121
                    messages.append(_tool_result_message(
122
                        tool_call_id=tool_call_id,
123
                        content=f"INFRA_ERROR: {e}",
124
                        is_error=True,
125
                    ))
126
127
                    await tools.send_event(
128
                        content=f"{name} infra error: {e}",
129
                        message_type="error",
130
                        metadata={"tool": name, "class": "infra"},
131
                    )
132
                    raise
133
134
        # Max iterations exceeded
135
        await tools.send_event(
136
            content=f"Exceeded max tool iterations ({MAX_TOOL_ITERS})",
137
            message_type="error",
138
        )
139
        raise RuntimeError("Tool loop exceeded max iterations")

1
from thenvoi.core.simple_adapter import SimpleAdapter
2
from thenvoi.core.protocols import AgentToolsProtocol
3
from thenvoi.core.types import PlatformMessage
4
5
class MyFrameworkAdapter(SimpleAdapter[MyHistoryType]):
6
    """Adapter for MyFramework."""
7
8
    def __init__(
9
        self,
10
        model: str = "gpt-4o",
11
        custom_section: str = "",
12
        history_converter: MyHistoryConverter | None = None,
13
    ):
14
        super().__init__(
15
            history_converter=history_converter or MyHistoryConverter()
16
        )
17
        self.model = model
18
        self.custom_section = custom_section
19
        self._system_prompt = ""
20
21
    async def on_started(self, agent_name: str, agent_description: str) -> None:
22
        """Called after platform connection established."""
23
        await super().on_started(agent_name, agent_description)
24
        self._system_prompt = render_system_prompt(
25
            agent_name=agent_name,
26
            agent_description=agent_description,
27
            custom_section=self.custom_section,
28
        )
29
30
    async def on_message(
31
        self,
32
        msg: PlatformMessage,
33
        tools: AgentToolsProtocol,
34
        history: MyHistoryType,
35
        participants_msg: str | None,
36
        *,
37
        is_session_bootstrap: bool,
38
        room_id: str,
39
    ) -> None:
40
        """Handle incoming message - implement your LLM interaction here."""
41
        # See patterns above
42
        ...
43
44
    async def on_cleanup(self, room_id: str) -> None:
45
        """Clean up when leaving a room."""
46
        ...

1
from thenvoi.core.protocols import HistoryConverter
2
3
# Define your history type
4
MyMessages = list[dict[str, Any]]  # or your framework's message type
5
6
class MyHistoryConverter(HistoryConverter[MyMessages]):
7
    """Convert platform history to MyFramework format."""
8
9
    def convert(self, raw: list[dict[str, Any]]) -> MyMessages:
10
        """
11
        Convert raw platform history.
12
13
        Each dict in raw has:
14
        - role: "user" or "assistant"
15
        - content: message content
16
        - sender_name: who sent it
17
        - sender_type: "User" or "Agent"
18
        - message_type: "text", "tool_call", "tool_result", etc.
19
        """
20
        messages = []
21
        for msg in raw:
22
            # Convert to your framework's format
23
            messages.append({
24
                "role": msg["role"],
25
                "content": msg["content"],
26
                # Add framework-specific fields...
27
            })
28
        return messages

1
# Get schemas in provider format - descriptions included automatically
2
tool_schemas = tools.get_tool_schemas("openai")  # or "anthropic"

1
from thenvoi.runtime.tools import get_tool_description
2
3
def convert_tools_to_my_framework(tools: AgentToolsProtocol) -> list[MyToolType]:
4
    """Convert AgentTools to MyFramework tool format."""
5
6
    # Create wrapper functions
7
    async def send_message_wrapper(content: str, mentions: list[str]) -> dict:
8
        return await tools.send_message(content, mentions)
9
10
    # Use centralized descriptions
11
    return [
12
        MyTool(
13
            name="thenvoi_send_message",
14
            description=get_tool_description("thenvoi_send_message"),
15
            func=send_message_wrapper,
16
        ),
17
        # ... other tools ...
18
    ]

1
from .my_framework import MyFrameworkAdapter
2
3
__all__ = [
4
    # ... existing adapters ...
5
    "MyFrameworkAdapter",
6
]

1
"""Minimal adapter using Pattern 2 (adapter manages tool loop)."""
2
3
from thenvoi.core.simple_adapter import SimpleAdapter
4
from thenvoi.core.protocols import AgentToolsProtocol
5
from thenvoi.core.types import PlatformMessage
6
from thenvoi.runtime.prompts import render_system_prompt
7
8
class MinimalAdapter(SimpleAdapter[list[dict]]):
9
    """
10
    Minimal adapter that manages its own message history.
11
12
    Uses history_converter=None to bypass platform history conversion,
13
    maintaining per-room state internally instead.
14
    """
15
16
    def __init__(self, api_key: str, model: str = "gpt-4o"):
17
        # No history converter - we manage history ourselves
18
        super().__init__(history_converter=None)
19
        self.api_key = api_key
20
        self.model = model
21
        self._system_prompt = ""
22
        self._room_messages: dict[str, list] = {}  # Per-room message history
23
24
    async def on_started(self, agent_name: str, agent_description: str) -> None:
25
        await super().on_started(agent_name, agent_description)
26
        self._system_prompt = render_system_prompt(
27
            agent_name=agent_name,
28
            agent_description=agent_description,
29
        )
30
31
    async def on_message(
32
        self,
33
        msg: PlatformMessage,
34
        tools: AgentToolsProtocol,
35
        history,  # Ignored - we manage our own history
36
        participants_msg: str | None,
37
        *,
38
        is_session_bootstrap: bool,
39
        room_id: str,
40
    ) -> None:
41
        # Initialize room on first message
42
        if is_session_bootstrap:
43
            self._room_messages[room_id] = [
44
                {"role": "system", "content": self._system_prompt}
45
            ]
46
47
        messages = self._room_messages[room_id]
48
49
        # Add user message
50
        messages.append({"role": "user", "content": msg.format_for_llm()})
51
52
        # Get tool schemas (sync method)
53
        tool_schemas = tools.get_tool_schemas("openai")
54
55
        # Tool execution loop
56
        while True:
57
            response = await self._call_llm(messages, tool_schemas)
58
59
            # Check if LLM wants to use tools
60
            if not response.get("tool_calls"):
61
                # No tools - add final assistant message and exit
62
                if response.get("content"):
63
                    messages.append({
64
                        "role": "assistant",
65
                        "content": response["content"],
66
                    })
67
                break
68
69
            # Append assistant response with tool calls
70
            messages.append({
71
                "role": "assistant",
72
                "content": response.get("content", ""),
73
                "tool_calls": response["tool_calls"],
74
            })
75
76
            # Execute tools and collect results
77
            for tool_call in response["tool_calls"]:
78
                result = await tools.execute_tool_call(
79
                    tool_call["name"],
80
                    tool_call["arguments"],
81
                )
82
                # Append tool result
83
                messages.append({
84
                    "role": "tool",
85
                    "tool_call_id": tool_call["id"],
86
                    "content": str(result),
87
                })
88
89
    async def on_cleanup(self, room_id: str) -> None:
90
        """Clean up room state when session ends."""
91
        self._room_messages.pop(room_id, None)

1
from thenvoi.testing import FakeAgentTools
2
3
async def test_my_adapter():
4
    adapter = MyFrameworkAdapter(model="gpt-4o")
5
    tools = FakeAgentTools()
6
7
    # Simulate a message
8
    await adapter.on_message(
9
        msg=PlatformMessage(id="1", content="Hello", sender_name="User"),
10
        tools=tools,
11
        history=[],
12
        participants_msg=None,
13
        is_session_bootstrap=True,
14
        room_id="room-1",
15
    )
16
17
    # Assert on tool calls
18
    assert tools.messages_sent  # LLM called thenvoi_send_message