Skip to main content

Framework Adapters

AG-Kit provides bidirectional adapters to seamlessly integrate with popular AI frameworks like LangChain and LlamaIndex. Convert AG-Kit tools to framework-specific tools and vice versa.

Overview

The adapter system enables:
  • AG-Kit → LangChain: Convert any AG-Kit tool to LangChain StructuredTool
  • AG-Kit → LlamaIndex: Convert any AG-Kit tool to LlamaIndex FunctionTool
  • LangChain → AG-Kit: Import LangChain tools into AG-Kit
  • LlamaIndex → AG-Kit: Import LlamaIndex tools into AG-Kit
  • Universal Compatibility: All AG-Kit tools (Code Executors, Bash, File System) work with both frameworks

Quick Start

LangChain Integration

from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool

# Create AG-Kit tool
executor = UnsafeLocalCodeExecutor()

# Convert to LangChain tool
langchain_tool = AGKitTool(tool=executor).as_langchain_tool()

# Use with LangChain
result = await langchain_tool.ainvoke({
    "language": "python",
    "code": "print('Hello from LangChain!')"
})

LlamaIndex Integration

from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool

# Create AG-Kit tool
executor = UnsafeLocalCodeExecutor()

# Convert to LlamaIndex tool
llamaindex_tool = AGKitTool(tool=executor).as_llamaindex_tool()

# Use with LlamaIndex
result = await llamaindex_tool.acall(
    language="python",
    code="print('Hello from LlamaIndex!')"
)

Installation

LangChain Support

pip install ag_kit_py langchain-core

LlamaIndex Support

pip install ag_kit_py llama-index-core

Supported Tools

All AG-Kit tools are compatible with both frameworks:
Tool CategoryToolsLangChainLlamaIndex
Code ExecutorsUnsafeLocalCodeExecutor, BuiltInCodeExecutor
Bash Toolscreate_bash_tool, create_multi_command_tool
File Systemcreate_read_tool, create_write_tool, create_edit_tool, etc.

LangChain Adapter

Converting AG-Kit Tools to LangChain

from ag_kit_py.tools import (
    UnsafeLocalCodeExecutor,
    create_bash_tool,
    create_read_tool,
)
from ag_kit_py.tools.bash import create_local_bash_context
from ag_kit_py.tools.fs import ExecutionContext, LocalFileOperator
from ag_kit_py.tools.adapters import AGKitTool

# Code Executor
executor = UnsafeLocalCodeExecutor()
lc_executor = AGKitTool(tool=executor).as_langchain_tool()

# Bash Tool
bash_context = create_local_bash_context()
bash_tool = create_bash_tool(bash_context)
lc_bash = AGKitTool(tool=bash_tool).as_langchain_tool()

# File System Tool
fs_context = ExecutionContext(
    working_directory="/workspace",
    fs_operator=LocalFileOperator()
)
read_tool = create_read_tool(fs_context)
lc_read = AGKitTool(tool=read_tool).as_langchain_tool()

# Use with LangChain agents
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

tools = [lc_executor, lc_bash, lc_read]
llm = ChatOpenAI(model="gpt-4")
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

Converting LangChain Tools to AG-Kit

from langchain_core.tools import StructuredTool
from ag_kit_py.tools.adapters import from_langchain

# Create LangChain tool
def calculate_area(length: float, width: float) -> float:
    """Calculate the area of a rectangle."""
    return length * width

langchain_tool = StructuredTool.from_function(
    func=calculate_area,
    name="area_calculator",
    description="Calculate the area of a rectangle"
)

# Convert to AG-Kit tool
agkit_tool = from_langchain(langchain_tool)

# Use as AG-Kit tool
result = await agkit_tool.invoke({"length": 5.0, "width": 3.0})
print(result.data)  # 15.0

LangChain with LangGraph

from langgraph.prebuilt import ToolNode
from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool

# Convert AG-Kit tools
executor = UnsafeLocalCodeExecutor()
langchain_tools = [AGKitTool(tool=executor).as_langchain_tool()]

# Create LangGraph ToolNode
tool_node = ToolNode(langchain_tools)

# Use in LangGraph workflow
from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.add_edge("agent", "tools")
workflow.add_edge("tools", END)

LlamaIndex Adapter

Converting AG-Kit Tools to LlamaIndex

from ag_kit_py.tools import (
    UnsafeLocalCodeExecutor,
    create_bash_tool,
    create_write_tool,
)
from ag_kit_py.tools.bash import create_local_bash_context
from ag_kit_py.tools.fs import ExecutionContext, LocalFileOperator
from ag_kit_py.tools.adapters import AGKitTool

# Code Executor
executor = UnsafeLocalCodeExecutor()
li_executor = AGKitTool(tool=executor).as_llamaindex_tool()

# Bash Tool
bash_context = create_local_bash_context()
bash_tool = create_bash_tool(bash_context)
li_bash = AGKitTool(tool=bash_tool).as_llamaindex_tool()

# File System Tool
fs_context = ExecutionContext(
    working_directory="/workspace",
    fs_operator=LocalFileOperator()
)
write_tool = create_write_tool(fs_context)
li_write = AGKitTool(tool=write_tool).as_llamaindex_tool()

# Use with LlamaIndex agents
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

tools = [li_executor, li_bash, li_write]
llm = OpenAI(model="gpt-4")
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

Converting LlamaIndex Tools to AG-Kit

from llama_index.core.tools import FunctionTool
from ag_kit_py.tools.adapters import from_llamaindex

# Create LlamaIndex tool
def calculate_volume(length: float, width: float, height: float) -> float:
    """Calculate the volume of a box."""
    return length * width * height

llamaindex_tool = FunctionTool.from_defaults(
    fn=calculate_volume,
    name="volume_calculator",
    description="Calculate the volume of a box"
)

# Convert to AG-Kit tool
agkit_tool = from_llamaindex(llamaindex_tool)

# Use as AG-Kit tool
result = await agkit_tool.invoke({
    "length": 5.0,
    "width": 3.0,
    "height": 2.0
})
print(result.data)  # 30.0

Bidirectional Conversion

Convert tools back and forth between frameworks:
from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool, from_langchain, from_llamaindex

# Start with AG-Kit tool
executor = UnsafeLocalCodeExecutor()

# Convert to LangChain
langchain_tool = AGKitTool(tool=executor).as_langchain_tool()

# Convert back to AG-Kit
agkit_tool_1 = from_langchain(langchain_tool)

# Convert to LlamaIndex
llamaindex_tool = AGKitTool(tool=executor).as_llamaindex_tool()

# Convert back to AG-Kit
agkit_tool_2 = from_llamaindex(llamaindex_tool)

# All tools work identically
result1 = await executor.invoke({"language": "python", "code": "print('test')"})
result2 = await agkit_tool_1.invoke({"language": "python", "code": "print('test')"})
result3 = await agkit_tool_2.invoke({"language": "python", "code": "print('test')"})

Complete Examples

Example 1: Code Executor with Both Frameworks

import asyncio
from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool

async def main():
    # Create AG-Kit tool
    executor = UnsafeLocalCodeExecutor()
    
    # Convert to LangChain
    lc_tool = AGKitTool(tool=executor).as_langchain_tool()
    result1 = await lc_tool.ainvoke({
        "language": "python",
        "code": "result = sum([1, 2, 3, 4, 5])\nprint(f'Sum: {result}')"
    })
    print(f"LangChain result: {result1}")
    
    # Convert to LlamaIndex
    li_tool = AGKitTool(tool=executor).as_llamaindex_tool()
    result2 = await li_tool.acall(
        language="python",
        code="result = sum([1, 2, 3, 4, 5])\nprint(f'Sum: {result}')"
    )
    print(f"LlamaIndex result: {result2}")

asyncio.run(main())

Example 2: Bash Tools with Both Frameworks

import asyncio
import tempfile
from ag_kit_py.tools import create_bash_tool, create_multi_command_tool
from ag_kit_py.tools.bash import create_local_bash_context
from ag_kit_py.tools.adapters import AGKitTool

async def main():
    # Create bash context
    context = create_local_bash_context(cwd=tempfile.gettempdir())
    
    # Single command tool
    bash_tool = create_bash_tool(context)
    lc_bash = AGKitTool(tool=bash_tool).as_langchain_tool()
    li_bash = AGKitTool(tool=bash_tool).as_llamaindex_tool()
    
    # Execute with LangChain
    result1 = await lc_bash.ainvoke({"command": "echo 'Hello from LangChain!'"})
    print(f"LangChain: {result1}")
    
    # Execute with LlamaIndex
    result2 = await li_bash.acall(command="echo 'Hello from LlamaIndex!'")
    print(f"LlamaIndex: {result2}")
    
    # Multi-command tool
    multi_tool = create_multi_command_tool(context)
    lc_multi = AGKitTool(tool=multi_tool).as_langchain_tool()
    li_multi = AGKitTool(tool=multi_tool).as_llamaindex_tool()
    
    # Execute multiple commands
    commands = ["echo 'Command 1'", "echo 'Command 2'", "echo 'Command 3'"]
    result3 = await lc_multi.ainvoke({"commands": commands})
    print(f"LangChain multi: {result3}")

asyncio.run(main())

Example 3: File System Tools with Both Frameworks

import asyncio
import tempfile
from ag_kit_py.tools.fs import (
    create_read_tool,
    create_write_tool,
    ExecutionContext,
    LocalFileOperator,
)
from ag_kit_py.tools.adapters import AGKitTool

async def main():
    with tempfile.TemporaryDirectory() as temp_dir:
        # Create FS context
        context = ExecutionContext(
            working_directory=temp_dir,
            fs_operator=LocalFileOperator()
        )
        
        # Create tools
        write_tool = create_write_tool(context)
        read_tool = create_read_tool(context)
        
        # Convert to LangChain
        lc_write = AGKitTool(tool=write_tool).as_langchain_tool()
        lc_read = AGKitTool(tool=read_tool).as_langchain_tool()
        
        # Write file with LangChain
        await lc_write.ainvoke({
            "file_path": "test.txt",
            "content": "Hello from LangChain!",
            "create_dirs": True
        })
        
        # Read file with LangChain
        content1 = await lc_read.ainvoke({"file_path": "test.txt"})
        print(f"LangChain read: {content1}")
        
        # Convert to LlamaIndex
        li_write = AGKitTool(tool=write_tool).as_llamaindex_tool()
        li_read = AGKitTool(tool=read_tool).as_llamaindex_tool()
        
        # Write file with LlamaIndex
        await li_write.acall(
            file_path="test2.txt",
            content="Hello from LlamaIndex!",
            create_dirs=True
        )
        
        # Read file with LlamaIndex
        content2 = await li_read.acall(file_path="test2.txt")
        print(f"LlamaIndex read: {content2}")

asyncio.run(main())

Example 4: Workflow Integration

import asyncio
import tempfile
from ag_kit_py.tools import UnsafeLocalCodeExecutor, create_bash_tool
from ag_kit_py.tools.bash import create_local_bash_context
from ag_kit_py.tools.fs import (
    create_read_tool,
    create_write_tool,
    ExecutionContext,
    LocalFileOperator,
)
from ag_kit_py.tools.adapters import AGKitTool

async def main():
    with tempfile.TemporaryDirectory() as temp_dir:
        # Setup contexts
        fs_context = ExecutionContext(
            working_directory=temp_dir,
            fs_operator=LocalFileOperator()
        )
        bash_context = create_local_bash_context(cwd=temp_dir)
        
        # Create and convert all tools to LangChain
        tools = [
            AGKitTool(tool=create_write_tool(fs_context)).as_langchain_tool(),
            AGKitTool(tool=create_read_tool(fs_context)).as_langchain_tool(),
            AGKitTool(tool=create_bash_tool(bash_context)).as_langchain_tool(),
            AGKitTool(tool=UnsafeLocalCodeExecutor()).as_langchain_tool(),
        ]
        
        # Workflow: Write → Read → Process → Execute
        print("1. Writing Python script...")
        await tools[0].ainvoke({
            "file_path": "script.py",
            "content": "numbers = [1, 2, 3, 4, 5]\nprint(f'Sum: {sum(numbers)}')",
            "create_dirs": True
        })
        
        print("2. Reading script...")
        content = await tools[1].ainvoke({"file_path": "script.py"})
        print(f"   Content length: {len(str(content))} chars")
        
        print("3. Listing files...")
        await tools[2].ainvoke({"command": "ls -la"})
        
        print("4. Executing script...")
        result = await tools[3].ainvoke({
            "language": "python",
            "code": "numbers = [1, 2, 3, 4, 5]\nprint(f'Sum: {sum(numbers)}')"
        })
        print(f"   Result: {result[:50]}...")
        
        print("\n✅ Workflow completed successfully!")

asyncio.run(main())

API Reference

AGKitTool

Wrapper class for converting AG-Kit tools to other frameworks.
from ag_kit_py.tools.adapters import AGKitTool

class AGKitTool:
    def __init__(self, tool: BaseTool):
        """
        Initialize adapter with an AG-Kit tool.
        
        Args:
            tool: Any AG-Kit BaseTool instance
        """
        ...
    
    def as_langchain_tool(self) -> StructuredTool:
        """
        Convert to LangChain StructuredTool.
        
        Returns:
            LangChain StructuredTool instance
        """
        ...
    
    def as_llamaindex_tool(self) -> FunctionTool:
        """
        Convert to LlamaIndex FunctionTool.
        
        Returns:
            LlamaIndex FunctionTool instance
        """
        ...

Conversion Functions

from ag_kit_py.tools.adapters import from_langchain, from_llamaindex

def from_langchain(langchain_tool: StructuredTool) -> BaseTool:
    """
    Convert LangChain tool to AG-Kit tool.
    
    Args:
        langchain_tool: LangChain StructuredTool instance
    
    Returns:
        AG-Kit BaseTool instance
    """
    ...

def from_llamaindex(llamaindex_tool: FunctionTool) -> BaseTool:
    """
    Convert LlamaIndex tool to AG-Kit tool.
    
    Args:
        llamaindex_tool: LlamaIndex FunctionTool instance
    
    Returns:
        AG-Kit BaseTool instance
    """
    ...

Error Handling

Both adapters preserve error handling behavior:
from ag_kit_py.tools import UnsafeLocalCodeExecutor
from ag_kit_py.tools.adapters import AGKitTool

executor = UnsafeLocalCodeExecutor()
lc_tool = AGKitTool(tool=executor).as_langchain_tool()

# Errors are properly propagated
try:
    result = await lc_tool.ainvoke({
        "language": "python",
        "code": "undefined_variable * 2"
    })
    # Check result for execution errors
    if "error" in str(result).lower():
        print("Execution error occurred")
except Exception as e:
    print(f"Tool error: {e}")

Best Practices

  1. Choose the right framework: Use LangChain for complex agent workflows, LlamaIndex for RAG applications
  2. Reuse contexts: Create contexts once and reuse for multiple tools
  3. Handle errors gracefully: Check tool results for errors
  4. Clean up resources: Properly dispose of file operators and bash contexts
  5. Test conversions: Verify tool behavior after conversion
  6. Use type hints: Leverage Python type hints for better IDE support

Performance Considerations

  • Conversion overhead: Minimal - adapters are lightweight wrappers
  • Execution performance: Identical to native AG-Kit tools
  • Memory usage: No significant overhead from adapters
  • Async support: Full async/await support in both frameworks

Troubleshooting

Common Issues

Tool not found after conversion
# Ensure tool is properly initialized
tool = UnsafeLocalCodeExecutor()
lc_tool = AGKitTool(tool=tool).as_langchain_tool()
print(lc_tool.name)  # Verify tool name
Schema validation errors
# Check tool schema
tool = UnsafeLocalCodeExecutor()
schema = tool.to_json_schema()
print(schema)  # Verify schema structure
Import errors
# Install required dependencies
pip install langchain-core  # For LangChain
pip install llama-index-core  # For LlamaIndex

Examples Repository

Find complete working examples at: