Tool Integration

1
from line.events import ToolCall, ToolResult
2
3
# Agent generates tool calls
4
class ToolCapableAgent(ReasoningNode):
5
    async def process_context(self, context):
6
        user_msg = context.get_latest_user_transcript_message()
7
        
8
        if "weather" in user_msg.lower():
9
            # Request weather information
10
            yield ToolCall(
11
                tool_name="get_weather",
12
                tool_args={"location": "New York", "units": "fahrenheit"}
13
            )
14
            
15
        # Continue processing after tools complete
16
        yield AgentResponse(content="Let me check that for you...")
17
18
# Tool execution handler
19
async def execute_weather_tool(tool_call):
20
    try:
21
        weather_data = await weather_api.get_current(
22
            location=tool_call.tool_args["location"],
23
            units=tool_call.tool_args["units"]
24
        )
25
        
26
        return ToolResult(
27
            tool_name=tool_call.tool_name,
28
            tool_args=tool_call.tool_args,
29
            result=weather_data,
30
            tool_call_id=tool_call.tool_call_id
31
        )
32
    except Exception as e:
33
        return ToolResult(
34
            tool_name=tool_call.tool_name,
35
            tool_args=tool_call.tool_args,
36
            error=str(e),
37
            tool_call_id=tool_call.tool_call_id
38
        )
39
40
# Route tool calls to handlers
41
bridge.on(ToolCall).map(execute_weather_tool).broadcast()
42
43
# Route results back to agent
44
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(tool_name="get_weather", tool_args={"location": "NYC"})
12
            
13
        if self.needs_calendar_info(user_input):
14
            yield ToolCall(tool_name="get_calendar", tool_args={"date": "today"})
15
            
16
        if self.needs_calculation(user_input):
17
            yield ToolCall(tool_name="calculate", tool_args={"expression": "2+2"})
18
            
19
        # Wait for all tool results, then respond
20
        yield AgentResponse(content="Let me gather that information...")
21
22
# Tool registry and dispatcher
23
class ToolDispatcher:
24
    def __init__(self):
25
        self.tools = {
26
            "get_weather": self.get_weather,
27
            "get_calendar": self.get_calendar,
28
            "calculate": self.calculate
29
        }
30
        
31
    async def execute_tool(self, tool_call):
32
        tool_func = self.tools.get(tool_call.tool_name)
33
        if not tool_func:
34
            return ToolResult(
35
                tool_name=tool_call.tool_name,
36
                tool_args=tool_call.tool_args,
37
                error=f"Unknown tool: {tool_call.tool_name}",
38
                tool_call_id=tool_call.tool_call_id
39
            )
40
            
41
        try:
42
            result = await tool_func(**tool_call.tool_args)
43
            return ToolResult(
44
                tool_name=tool_call.tool_name,
45
                tool_args=tool_call.tool_args,
46
                result=result,
47
                tool_call_id=tool_call.tool_call_id
48
            )
49
        except Exception as e:
50
            return ToolResult(
51
                tool_name=tool_call.tool_name,
52
                tool_args=tool_call.tool_args,
53
                error=str(e),
54
                tool_call_id=tool_call.tool_call_id
55
            )
56
57
dispatcher = ToolDispatcher()
58
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 = [e for e in context.events if isinstance(e, ToolResult)]
5
        
6
        if tool_results:
7
            # Process tool results
8
            for result in tool_results:
9
                if result.success:
10
                    yield AgentResponse(
11
                        content=f"Based on {result.tool_name}, the result is: {result.result_str}"
12
                    )
13
                else:
14
                    yield AgentResponse(
15
                        content=f"I encountered an error with {result.tool_name}: {result.error}"
16
                    )
17
        else:
18
            # Normal conversation flow
19
            user_msg = context.get_latest_user_transcript_message()
20
            
21
            # Check if we need to call tools
22
            if self.requires_external_data(user_msg):
23
                yield ToolCall(tool_name="search", tool_args={"query": user_msg})
24
            else:
25
                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(tool_name="end_call", tool_args={"reason": "User requested"})
6
7
# Transfer call tool  
8
yield ToolCall(tool_name="transfer_call", tool_args={
9
    "destination": "+15551234567",
10
    "reason": "Technical support needed"
11
})
12
13
# Agent handoff tools (following naming convention)
14
yield ToolCall(tool_name="transfer_to_billing", tool_args={
15
    "reason": "Customer has billing questions"
16
})

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
)