Tool Integration

1
from line.events import ToolCall, ToolResult
2
3
4
# Agent generates tool calls
5
class ToolCapableAgent(ReasoningNode):
6
    async def process_context(self, context):
7
        user_msg = context.get_latest_user_transcript_message()
8
9
        if "weather" in user_msg.lower():
10
            # Request weather information
11
            yield ToolCall(
12
                tool_name="get_weather",
13
                tool_args={
14
                    "location": "New York",
15
                    "units": "fahrenheit",
16
                },
17
            )
18
19
        # Continue processing after tools complete
20
        yield AgentResponse(content="Let me check that for you...")
21
22
23
# Tool execution handler
24
async def execute_weather_tool(tool_call):
25
    try:
26
        weather_data = await weather_api.get_current(
27
            location=tool_call.tool_args["location"],
28
            units=tool_call.tool_args["units"],
29
        )
30
31
        return ToolResult(
32
            tool_name=tool_call.tool_name,
33
            tool_args=tool_call.tool_args,
34
            result=weather_data,
35
            tool_call_id=tool_call.tool_call_id,
36
        )
37
    except Exception as e:
38
        return ToolResult(
39
            tool_name=tool_call.tool_name,
40
            tool_args=tool_call.tool_args,
41
            error=str(e),
42
            tool_call_id=tool_call.tool_call_id,
43
        )
44
45
46
# Route tool calls to handlers
47
bridge.on(ToolCall).map(execute_weather_tool).broadcast()
48
49
# Route results back to agent
50
bridge.on(ToolResult).map(node.add_event)

1
class MultiToolAgent(ReasoningNode):
2
    def __init__(self, tools_config):
3
        super().__init__()
4
        self.tools = tools_config
5
6
    async def process_context(self, context):
7
        user_input = context.get_latest_user_transcript_message()
8
9
        # Analyze what tools are needed
10
        if self.needs_weather_info(user_input):
11
            yield ToolCall(
12
                tool_name="get_weather", tool_args={"location": "NYC"}
13
            )
14
15
        if self.needs_calendar_info(user_input):
16
            yield ToolCall(
17
                tool_name="get_calendar", tool_args={"date": "today"}
18
            )
19
20
        if self.needs_calculation(user_input):
21
            yield ToolCall(
22
                tool_name="calculate", tool_args={"expression": "2+2"}
23
            )
24
25
        # Wait for all tool results, then respond
26
        yield AgentResponse(
27
            content="Let me gather that information..."
28
        )
29
30
31
# Tool registry and dispatcher
32
class ToolDispatcher:
33
    def __init__(self):
34
        self.tools = {
35
            "get_weather": self.get_weather,
36
            "get_calendar": self.get_calendar,
37
            "calculate": self.calculate,
38
        }
39
40
    async def execute_tool(self, tool_call):
41
        tool_func = self.tools.get(tool_call.tool_name)
42
        if not tool_func:
43
            return ToolResult(
44
                tool_name=tool_call.tool_name,
45
                tool_args=tool_call.tool_args,
46
                error=f"Unknown tool: {tool_call.tool_name}",
47
                tool_call_id=tool_call.tool_call_id,
48
            )
49
50
        try:
51
            result = await tool_func(**tool_call.tool_args)
52
            return ToolResult(
53
                tool_name=tool_call.tool_name,
54
                tool_args=tool_call.tool_args,
55
                result=result,
56
                tool_call_id=tool_call.tool_call_id,
57
            )
58
        except Exception as e:
59
            return ToolResult(
60
                tool_name=tool_call.tool_name,
61
                tool_args=tool_call.tool_args,
62
                error=str(e),
63
                tool_call_id=tool_call.tool_call_id,
64
            )
65
66
67
dispatcher = ToolDispatcher()
68
bridge.on(ToolCall).map(dispatcher.execute_tool).broadcast()

1
class ToolAwareAgent(ReasoningNode):
2
    async def process_context(self, context):
3
        # Check for tool results in context
4
        tool_results = [
5
            e for e in context.events if isinstance(e, ToolResult)
6
        ]
7
8
        if tool_results:
9
            # Process tool results
10
            for result in tool_results:
11
                if result.success:
12
                    yield AgentResponse(
13
                        content=f"Based on {result.tool_name}, the result is: {result.result_str}"
14
                    )
15
                else:
16
                    yield AgentResponse(
17
                        content=f"I encountered an error with {result.tool_name}: {result.error}"
18
                    )
19
        else:
20
            # Normal conversation flow
21
            user_msg = context.get_latest_user_transcript_message()
22
23
            # Check if we need to call tools
24
            if self.requires_external_data(user_msg):
25
                yield ToolCall(
26
                    tool_name="search", tool_args={"query": user_msg}
27
                )
28
            else:
29
                yield AgentResponse(content="How can I help you?")

1
# Built-in system tools for common operations
2
from line.tools import system_tools
3
4
# End call tool
5
yield ToolCall(
6
    tool_name="end_call", tool_args={"reason": "User requested"}
7
)
8
9
# Transfer call tool
10
yield ToolCall(
11
    tool_name="transfer_call",
12
    tool_args={
13
        "destination": "+15551234567",
14
        "reason": "Technical support needed",
15
    },
16
)
17
18
# Agent handoff tools (following naming convention)
19
yield ToolCall(
20
    tool_name="transfer_to_billing",
21
    tool_args={"reason": "Customer has billing questions"},
22
)

1
ToolResult(
2
    tool_name="unknown_tool",
3
    error="Tool 'unknown_tool' not found",
4
    tool_call_id=tool_call_id,
5
)

1
ToolResult(
2
    tool_name="database_query",
3
    error="Connection timeout to database",
4
    tool_call_id=tool_call_id,
5
)

1
ToolResult(
2
    tool_name="weather_api",
3
    error="Invalid location parameter: 'xyz123'",
4
    tool_call_id=tool_call_id,
5
)