Creating Framework Integrations

async def on_message(
    self, msg, tools, history, participants_msg, contacts_msg,
    *, is_session_bootstrap: bool, room_id: str,
):
    # Convert platform tools to LangChain format
    langchain_tools = agent_tools_to_langchain(tools)
    # Create graph with tools - graph handles tool execution internally
    graph = create_react_agent(llm, langchain_tools, checkpointer)
    # Stream events - LLM decides when to call band_send_message
    async for event in graph.astream_events({"messages": messages}, ...):
        await self._handle_stream_event(event, room_id, tools)

async def on_message(
    self, msg, tools, history, participants_msg, contacts_msg,
    *, is_session_bootstrap: bool, room_id: str,
):
    # Get tool schemas in Anthropic format (sync method)
    tool_schemas = tools.get_tool_schemas(
        "anthropic",
        capabilities=self.features.capabilities,
    )
    # Tool execution loop
    while True:
        # Call LLM with tools
        response = await self.client.messages.create(
            model=self.model,
            messages=messages,
            tools=tool_schemas,
        )
        # Check if LLM wants to use tools
        if response.stop_reason != "tool_use":
            break  # LLM is done
        # IMPORTANT: Append assistant response (with tool_use blocks) to history
        messages.append({
            "role": "assistant",
            "content": response.content,  # Contains ToolUseBlock(s)
        })
        # Execute tool calls and collect results
        tool_results = []
        for block in response.content:
            if isinstance(block, ToolUseBlock):
                result = await tools.execute_tool_call(
                    block.name,  # e.g., "band_send_message"
                    block.input  # e.g., {"content": "Hello!", "mentions": ["User"]}
                )
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                })
        # IMPORTANT: Append tool results to history
        messages.append({
            "role": "user",
            "content": tool_results,
        })

import json
from band.core import AgentToolsProtocol
async def _handle_stream_event(
    self,
    event: dict,
    room_id: str,
    tools: AgentToolsProtocol,
) -> None:
    """Handle streaming events from framework."""
    event_type = event.get("event")
    if event_type == "on_tool_start":
        tool_name = event.get("name", "unknown")
        await tools.send_event(
            content=json.dumps(event, default=str),
            message_type="tool_call",
        )
    elif event_type == "on_tool_end":
        tool_name = event.get("name", "unknown")
        await tools.send_event(
            content=json.dumps(event, default=str),
            message_type="tool_result",
        )

async def _process_tool_calls(
    self,
    response: "Message",  # anthropic.types.Message
    tools: AgentToolsProtocol,
) -> list[dict]:
    """Execute tool calls and report events."""
    results = []
    for block in response.content:
        if not isinstance(block, ToolUseBlock):
            continue
        # Report tool call
        await tools.send_event(
            content=f"Calling {block.name}",
            message_type="tool_call",
            metadata={"tool": block.name, "input": block.input},
        )
        # Execute tool
        try:
            result = await tools.execute_tool_call(block.name, block.input)
            is_error = False
        except Exception as e:
            result = f"Error: {e}"
            is_error = True
        # Report result
        await tools.send_event(
            content=f"Result: {result}",
            message_type="tool_result",
            metadata={"tool": block.name, "is_error": is_error},
        )
        results.append({"tool_use_id": block.id, "content": str(result)})
    return results

async def on_message(
    self, msg, tools, history, participants_msg, contacts_msg,
    *, is_session_bootstrap: bool, room_id: str,
):
    try:
        response = await self._call_llm(messages, tool_schemas)
        # ... process response ...
    except Exception as e:
        # Report error to platform
        await tools.send_event(
            content=f"Error: {e}",
            message_type="error",
        )
        raise  # Re-raise so message is marked as failed

import json
from typing import Any
MAX_TOOL_ITERS = 10
class ToolRecoverableError(Exception):
    """Errors the LLM can reasonably react to (bad args, not found, permission, etc)."""
class ToolInfraError(Exception):
    """Errors that indicate runtime/infrastructure problems (timeouts, 5xx, auth, etc)."""
def _maybe_json_loads(x: Any) -> Any:
    """Parse JSON string if needed; some LLM APIs return arguments as strings."""
    if isinstance(x, str):
        try:
            return json.loads(x)
        except json.JSONDecodeError:
            return x
    return x
def _tool_result_message(*, tool_call_id: str, content: str, is_error: bool) -> dict[str, Any]:
    """Build OpenAI-style tool result message."""
    prefix = "ERROR: " if is_error else ""
    return {
        "role": "tool",
        "tool_call_id": tool_call_id,
        "content": f"{prefix}{content}",
    }
class ManualToolLoopAdapter:
    def __init__(self, client, model: str):
        self.client = client
        self.model = model
    async def on_message(
        self, msg, tools, history, participants_msg, contacts_msg, *, is_session_bootstrap: bool, room_id: str
    ) -> None:
        messages: list[dict[str, Any]] = list(history)
        messages.append({"role": "user", "content": msg.format_for_llm()})
        tool_schemas = tools.get_tool_schemas("openai")
        for i in range(MAX_TOOL_ITERS):
            # 1) Call the LLM
            try:
                resp = await self.client.responses.create(
                    model=self.model,
                    input=messages,
                    tools=tool_schemas,
                )
            except Exception as e:
                await tools.send_event(content=f"LLM call failed: {e}", message_type="error")
                raise
            # 2) Extract assistant content + tool calls
            assistant_content = getattr(resp, "output_text", None) or ""
            tool_calls = getattr(resp, "tool_calls", None) or []
            # 3) No tool calls = done
            if not tool_calls:
                if assistant_content:
                    messages.append({"role": "assistant", "content": assistant_content})
                return
            # 4) Append assistant message with tool calls
            messages.append({
                "role": "assistant",
                "content": assistant_content,
                "tool_calls": tool_calls,
            })
            # 5) Execute tool calls
            for tc in tool_calls:
                name = tc["name"]
                tool_call_id = tc["id"]
                args = _maybe_json_loads(tc.get("arguments", {}))
                await tools.send_event(
                    content=f"Calling {name}",
                    message_type="tool_call",
                    metadata={"tool": name, "input": args},
                )
                try:
                    result = await tools.execute_tool_call(name, args)
                    await tools.send_event(
                        content=f"{name} OK",
                        message_type="tool_result",
                        metadata={"tool": name, "is_error": False},
                    )
                    messages.append(_tool_result_message(
                        tool_call_id=tool_call_id,
                        content=str(result),
                        is_error=False,
                    ))
                except ToolRecoverableError as e:
                    # Recoverable: give LLM the error so it can decide what to do
                    await tools.send_event(
                        content=f"{name} recoverable error: {e}",
                        message_type="tool_result",
                        metadata={"tool": name, "is_error": True, "class": "recoverable"},
                    )
                    messages.append(_tool_result_message(
                        tool_call_id=tool_call_id,
                        content=str(e),
                        is_error=True,
                    ))
                    # Do NOT raise — let the loop continue so LLM can react
                except Exception as e:
                    # Infra failure: append result for context, then fail
                    messages.append(_tool_result_message(
                        tool_call_id=tool_call_id,
                        content=f"INFRA_ERROR: {e}",
                        is_error=True,
                    ))
                    await tools.send_event(
                        content=f"{name} infra error: {e}",
                        message_type="error",
                        metadata={"tool": name, "class": "infra"},
                    )
                    raise
        # Max iterations exceeded
        await tools.send_event(
            content=f"Exceeded max tool iterations ({MAX_TOOL_ITERS})",
            message_type="error",
        )
        raise RuntimeError("Tool loop exceeded max iterations")

from typing import ClassVar, Unpack
from band import Capability, Emit, FeatureKwargs
from band.core import AgentToolsProtocol, SimpleAdapter
from band.core.types import PlatformMessage
class MyFrameworkAdapter(SimpleAdapter[MyHistoryType]):
    """Adapter for MyFramework."""
    # Declare what this adapter actually implements. The base constructor
    # raises BandConfigError for any emit or capability value outside these
    # sets, so a caller learns immediately rather than at run time.
    SUPPORTED_EMIT: ClassVar[frozenset[Emit]] = frozenset({Emit.TOOL_CALLS})
    # Capability.MEMORY, Capability.CONTACTS and Capability.FILES are the
    # declarable capabilities. Name only the ones you have wired: FILES needs
    # the room file tools, and reading a file can return an image, so an
    # adapter declaring it has to pass image content through to the model.
    SUPPORTED_CAPABILITIES: ClassVar[frozenset[Capability]] = frozenset()
    def __init__(
        self,
        model: str = "gpt-4o",
        custom_section: str = "",
        history_converter: MyHistoryConverter | None = None,
        **features: Unpack[FeatureKwargs],
    ):
        super().__init__(
            history_converter=history_converter or MyHistoryConverter(),
            **features,
        )
        self.model = model
        self.custom_section = custom_section
        self._system_prompt = ""
    async def on_started(self, agent_name: str, agent_description: str) -> None:
        """Called after agent metadata is fetched, before the WebSocket connects."""
        await super().on_started(agent_name, agent_description)
        self._system_prompt = render_system_prompt(
            agent_name=agent_name,
            agent_description=agent_description,
            custom_section=self.custom_section,
        )
    async def on_message(
        self,
        msg: PlatformMessage,
        tools: AgentToolsProtocol,
        history: MyHistoryType,
        participants_msg: str | None,
        contacts_msg: str | None,
        *,
        is_session_bootstrap: bool,
        room_id: str,
    ) -> None:
        """Handle incoming message - implement your LLM interaction here."""
        # Gate reporting on what the caller actually asked for, e.g.:
        # if Emit.TOOL_CALLS in self.features.emit: await tools.send_event(...)
        # See patterns above
        ...
    async def on_cleanup(self, room_id: str) -> None:
        """Clean up when leaving a room."""
        ...

from band.core import HistoryConverter
# Define your history type
MyMessages = list[dict[str, Any]]  # or your framework's message type
class MyHistoryConverter(HistoryConverter[MyMessages]):
    """Convert platform history to MyFramework format."""
    def convert(self, raw: list[dict[str, Any]]) -> MyMessages:
        """
        Convert raw platform history.
        Each dict in raw has:
        - role: "user" or "assistant"
        - content: message content
        - sender_name: who sent it
        - sender_type: "User" or "Agent"
        - message_type: "text", "tool_call", "tool_result", etc.
        """
        messages = []
        for msg in raw:
            # Convert to your framework's format
            messages.append({
                "role": msg["role"],
                "content": msg["content"],
                # Add framework-specific fields...
            })
        return messages

from band import Capability
# Get schemas in provider format - descriptions included automatically
tool_schemas = tools.get_tool_schemas("openai")  # or "anthropic"
# The same call with the memory tools advertised as well. `capabilities`
# replaces the default instead of adding to it, so name contacts again to keep
# the contact tools the default would have given you.
tool_schemas_with_memory = tools.get_tool_schemas(
    "openai",
    capabilities=frozenset({Capability.MEMORY, Capability.CONTACTS}),
)

from band.runtime.tools import get_tool_description
def convert_tools_to_my_framework(tools: AgentToolsProtocol) -> list[MyToolType]:
    """Convert AgentTools to MyFramework tool format."""
    # Create wrapper functions
    async def send_message_wrapper(content: str, mentions: list[str]) -> dict:
        return await tools.send_message(content, mentions)
    # Use centralized descriptions
    return [
        MyTool(
            name="band_send_message",
            description=get_tool_description("band_send_message"),
            func=send_message_wrapper,
        ),
        # ... other tools ...
    ]

"""Minimal adapter using Pattern 2 (adapter manages tool loop)."""
from band.core import AgentToolsProtocol, SimpleAdapter
from band.core.types import PlatformMessage
from band.runtime.prompts import render_system_prompt
class MinimalAdapter(SimpleAdapter[list[dict]]):
    """
    Minimal adapter that manages its own message history.
    Uses history_converter=None to bypass platform history conversion,
    maintaining per-room state internally instead.
    """
    def __init__(self, api_key: str, model: str = "gpt-4o"):
        # No history converter - we manage history ourselves
        super().__init__(history_converter=None)
        self.api_key = api_key
        self.model = model
        self._system_prompt = ""
        self._room_messages: dict[str, list] = {}  # Per-room message history
    async def on_started(self, agent_name: str, agent_description: str) -> None:
        await super().on_started(agent_name, agent_description)
        self._system_prompt = render_system_prompt(
            agent_name=agent_name,
            agent_description=agent_description,
        )
    async def on_message(
        self,
        msg: PlatformMessage,
        tools: AgentToolsProtocol,
        history,  # Ignored - we manage our own history
        participants_msg: str | None,
        contacts_msg: str | None,
        *,
        is_session_bootstrap: bool,
        room_id: str,
    ) -> None:
        # Initialize room on first message
        if is_session_bootstrap:
            self._room_messages[room_id] = [
                {"role": "system", "content": self._system_prompt}
            ]
        messages = self._room_messages[room_id]
        # Add user message
        messages.append({"role": "user", "content": msg.format_for_llm()})
        # Get tool schemas (sync method)
        tool_schemas = tools.get_tool_schemas(
            "openai",
            capabilities=self.features.capabilities,
        )
        # Tool execution loop
        while True:
            response = await self._call_llm(messages, tool_schemas)
            # Check if LLM wants to use tools
            if not response.get("tool_calls"):
                # No tools - add final assistant message and exit
                if response.get("content"):
                    messages.append({
                        "role": "assistant",
                        "content": response["content"],
                    })
                break
            # Append assistant response with tool calls
            messages.append({
                "role": "assistant",
                "content": response.get("content", ""),
                "tool_calls": response["tool_calls"],
            })
            # Execute tools and collect results
            for tool_call in response["tool_calls"]:
                result = await tools.execute_tool_call(
                    tool_call["name"],
                    tool_call["arguments"],
                )
                # Append tool result
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": str(result),
                })
    async def on_cleanup(self, room_id: str) -> None:
        """Clean up room state when session ends."""
        self._room_messages.pop(room_id, None)

from datetime import datetime, timezone
from band.core.types import PlatformMessage
from band.testing import FakeAgentTools
async def test_my_adapter():
    adapter = MyFrameworkAdapter(model="gpt-4o")
    tools = FakeAgentTools()
    # Simulate a message
    await adapter.on_message(
        msg=PlatformMessage(
            id="1",
            room_id="room-1",
            content="Hello",
            sender_id="user-1",
            sender_type="User",
            sender_name="User",
            message_type="text",
            metadata={},
            created_at=datetime.now(timezone.utc),
        ),
        tools=tools,
        history=[],
        participants_msg=None,
        contacts_msg=None,
        is_session_bootstrap=True,
        room_id="room-1",
    )
    # Assert on tool calls
    assert tools.messages_sent  # LLM called band_send_message