Agent Integration - AG-Kit Documentation

Learn how to integrate tools with AG-Kit agents to create powerful AI workflows that can interact with the external world through a comprehensive toolkit.

Basic Agent Integration

Simple Tool Integration

The most basic form of agent integration involves creating a custom tool and providing it to an agent. This allows the agent to perform specific operations beyond its base language capabilities.

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool as langchain_tool
from pydantic import BaseModel
import os

# Define tool schema
class CalculatorInput(BaseModel):
    operation: str
    a: float
    b: float

# Create a simple tool
@langchain_tool
def calculator(operation: str, a: float, b: float) -> dict:
    """Perform basic mathematical calculations"""
    if operation == 'divide' and b == 0:
        raise ValueError('Division by zero')
    
    operations = {'add': a + b, 'subtract': a - b, 'multiply': a * b, 'divide': a / b}
    return {'result': operations[operation], 'operation': operation, 'operands': [a, b]}

# Create model with tools
model = ChatOpenAI(model='gpt-4', api_key=os.getenv('OPENAI_API_KEY'))
model_with_tools = model.bind_tools([calculator])

# Create LangGraph workflow
workflow = StateGraph()
workflow.add_node('agent', lambda state: {
    'messages': [model_with_tools.invoke(state['messages'])]
})
workflow.add_edge(START, 'agent')
workflow.add_edge('agent', END)

graph = workflow.compile()

Key components of agent integration:

Model Provider: Configure the LLM provider (OpenAI, Anthropic, etc.)
Tool Definition: Create tools with clear schemas and error handling
Agent Configuration: Set up the agent with instructions, tools, and model settings
Execution: Run the agent with natural language inputs

Multiple Tool Integration

For more complex workflows, you can integrate multiple tools and toolkits with a single agent. This creates powerful assistants capable of handling diverse tasks across different domains.

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from agkit.tools import (
    FilesystemToolkit,
    BuiltInCodeExecutor,
    create_bash_tool,
    LocalFileOperator
)
from agkit.tools.adapters import AGKitTool
import os

# Set up file system context
fs_context = {
    'working_directory': os.getcwd(),
    'fs_operator': LocalFileOperator()
}

# Create filesystem toolkit
filesystem_toolkit = FilesystemToolkit(
    name='filesystem-toolkit',
    context=fs_context
)

# Convert AG-Kit tools to LangChain tools
agkit_tools = [
    *filesystem_toolkit.get_tools(),
    BuiltInCodeExecutor(),
    create_bash_tool(fs_context),
    calculator_tool
]
tools = [AGKitTool(tool=t).as_langchain_tool() for t in agkit_tools]

# Create model with tools
model = ChatOpenAI(model='gpt-4')
model_with_tools = model.bind_tools(tools)

# Create LangGraph workflow
workflow = StateGraph()
workflow.add_node('agent', lambda state: {
    'messages': [model_with_tools.invoke(state['messages'])]
})
workflow.add_edge(START, 'agent')
workflow.add_edge('agent', END)

graph = workflow.compile()

This approach allows agents to:

Read and write files using the filesystem toolkit
Execute code in secure sandboxed environments
Run command line operations for system interactions
Perform calculations with custom tools

Advanced Integration Patterns

Intelligent Tool Selection

Advanced agents can be configured with multiple specialized tools and intelligent selection strategies. The agent will automatically choose the most appropriate tool based on the task context and user requirements.

# Create specialized tools for different data formats
data_processing_tools = [
    tool(
        lambda file_path, operation: {
            'success': True, 
            'data': {'processed': True, 'operation': operation}
        },
        name='csv_processor',
        description='Process CSV data files and perform data analysis',
        schema=CSVProcessorInput
    ),
    tool(
        lambda url, method='GET', headers=None: {
            'success': True, 
            'data': {'status': 200, 'response': 'API response'}
        },
        name='http_client',
        description='Make HTTP requests to external APIs',
        schema=HTTPClientInput
    )
]

# Create agent with multiple specialized tools
smart_agent = # Agent initialization
    # ...
    tools=[*data_processing_tools, *filesystem_toolkit.get_tools()],
    # ...

Benefits of intelligent tool selection:

Automatic tool routing based on task requirements
Contextual decision making for optimal tool usage
Extensible architecture for adding new specialized tools
Clear reasoning about tool selection choices

Error Handling and Resilience

Graceful Error Recovery

Building resilient agents that can handle tool failures gracefully is crucial for production applications. AG-Kit provides several mechanisms for error handling and recovery.

# Create resilient agent with error handling
resilient_agent = # Agent initialization
    # ...
    tools=[
        *filesystem_toolkit.get_tools(),
        BuiltInCodeExecutor()
    ],
    # control_flow={'error_retry_limit': 2, 'max_steps': 10}

# Handle errors in usage
try:
    response = await resilient_agent.run('Process this complex request')
    print('Success:', response)
except Exception as error:
    print(f'Agent error: {error}')
    # Implement fallback strategies or user notification

Input Validation and Security

Implementing proper input validation and security measures in tools is essential for safe agent operations, especially when dealing with file systems or external APIs.

async def validated_file_processor(file_path: str, operation: str):
    try:
        # Pre-execution validation
        if not os.path.exists(file_path):
            return {
                'success': False,
                'error': 'File does not exist',
                'error_type': 'validation'
            }
        
        stats = os.stat(file_path)
        if stats.st_size > 10 * 1024 * 1024:  # 10MB limit
            return {
                'success': False,
                'error': 'File too large (max 10MB)',
                'error_type': 'validation'
            }
        
        # Execute operation with proper error handling
        operations = {
            'read': lambda: {'content': open(file_path).read()},
            'analyze': lambda: {
                'lines': len(open(file_path).readlines()),
                'characters': len(open(file_path).read()),
                'words': len(open(file_path).read().split())
            },
            'transform': lambda: {'message': 'Transform operation completed'}
        }
        
        result = operations[operation]()
        
        # Post-execution validation
        if not result or not isinstance(result, dict):
            return {
                'success': False,
                'error': 'Invalid processing result',
                'error_type': 'execution'
            }
        
        return {'success': True, 'data': result}
        
    except Exception as error:
        return {
            'success': False,
            'error': str(error),
            'error_type': 'execution'
        }

validated_tool = tool(
    validated_file_processor,
    name='validated_file_processor',
    description='Process files with comprehensive validation and security checks',
    schema=ValidatedFileProcessorInput
)

# Use validated tool with agent
secure_agent = # Agent initialization
    # ...
    tools=[validated_tool],
    # ...

Key validation practices:

Input sanitization to prevent path traversal attacks
File size limits to prevent resource exhaustion
Pre and post-execution validation for data integrity
Structured error responses with appropriate error types

Streaming and Real-time Integration

Streaming Tool Responses

For long-running operations or real-time data processing, AG-Kit supports streaming responses that provide progress updates and intermediate results to users.

import asyncio
from typing import AsyncIterator

# Create streaming tool for long-running operations
async def streaming_processor(dataset: str, batch_size: int = 100):
    async def stream_progress():
        for progress in [25, 50, 75, 100]:
            await asyncio.sleep(1)
            yield {
                'progress': progress,
                'message': f'Processing batch {progress // 25}',
                'timestamp': datetime.now().isoformat()
            }
    
    return {
        'success': True,
        'stream': stream_progress(),
        'data': {
            'message': 'Streaming processing started',
            'batch_size': batch_size,
            'estimated_duration': '4 seconds'
        }
    }

streaming_tool = tool(
    streaming_processor,
    name='streaming_processor',
    description='Process large datasets with real-time progress updates',
    schema=StreamingProcessorInput
)

# Create streaming agent
streaming_agent = # Agent initialization
    # ...
    tools=[streaming_tool],
    # ...

# Usage with streaming enabled
response = await streaming_agent.run(
    'Process the large dataset with real-time progress updates',
    # state, options
)

# Handle streaming response
if hasattr(response, 'stream'):
    async for chunk in response.stream:
        print(f'Streaming chunk: {chunk}')
        # Update UI with progress information

Streaming capabilities enable:

Real-time progress tracking for long-running operations
Intermediate result delivery for better user experience
Responsive UI updates during processing
Early error detection and handling

Performance Optimization

Parallel Tool Execution

AG-Kit agents can automatically execute independent tool calls in parallel, significantly improving performance for operations that don’t depend on each other.

# Create agent with parallel execution capabilities
parallel_agent = # Agent initialization
    # ...
    tools=[
        calculator_tool,
        *filesystem_toolkit.get_tools()
    ],
    # ...

# The agent will automatically execute independent tool calls in parallel
response = await parallel_agent.run(
    'Calculate 15 * 23, read the config.json file, and check if package.json exists'
)

Performance benefits:

Reduced execution time for independent operations
Better resource utilization through concurrent processing
Improved user experience with faster response times
Automatic optimization without manual coordination

Next Steps

Built-in Tools

Explore comprehensive built-in tools

Custom Tools

Build specialized tools for your agents

API Reference

Detailed API documentation

Documentation Index

​Basic Agent Integration

​Simple Tool Integration

​Multiple Tool Integration

​Advanced Integration Patterns

​Intelligent Tool Selection

​Error Handling and Resilience

​Graceful Error Recovery

​Input Validation and Security

​Streaming and Real-time Integration

​Streaming Tool Responses

​Performance Optimization

​Parallel Tool Execution

​Next Steps