Tool Provider
Organize and group tools from multiple applications using tool providers.
Tool Provider
The tool_provider parameter allows you to customize how tools are provided to the agent. By default, CUGA uses CombinedToolProvider internally. For advanced use cases, you can implement a custom tool provider or use one of the built-in implementations.
Why Import First?
When working with multiple applications and tools, you need to import the tool provider first to properly define and group tools by application. This allows you to:
- Organize tools by application: Group related tools together (e.g., all Slack tools, all GitHub tools)
- Manage tool namespaces: Avoid naming conflicts when tools from different apps have similar names
- Control tool visibility: Selectively load tools from specific applications
- Enable multi-app workflows: Use tools from multiple services in a single agent execution
By importing and configuring the tool provider before creating the agent, you establish the tool structure that the agent will use throughout its execution.
ToolProviderInterface
All tool providers implement the ToolProviderInterface abstract class, which defines the following interface:
Prop
Type
AppDefinition Schema
Applications are represented using the AppDefinition model:
Prop
Type
CombinedToolProvider
The CombinedToolProvider is the recommended provider for production use. It combines tools from both runtime tracker tools and the MCP registry, allowing you to execute tools against the registry while also supporting local runtime tools.
Note: The
CombinedToolProvideris used internally when you startcuga start demo. It requires the registry to be running and properly configured. To understand how the registry works and how to set it up, see the registry documentation.
How It Works
The CombinedToolProvider uses a two-tier approach:
- Runtime Tools (Priority): First checks the activity tracker for runtime tools that were dynamically added during execution
- Registry Tools (Fallback): Falls back to the MCP registry to load tools from OpenAPI specs or MCP servers
This allows the agent to execute tools against the registry while maintaining the flexibility to use locally defined runtime tools.
Executing Tools Against Registry
When tools are loaded from the registry, they are executed through the registry's execution layer. This means:
- OpenAPI Tools: Tools defined in OpenAPI specifications are executed via HTTP requests to the specified endpoints
- MCP Tools: Tools from MCP servers are executed through the MCP protocol
The registry handles authentication, request formatting, and response parsing automatically.
Usage
Prerequisite: The registry must be running before using
CombinedToolProvider. See the registry documentation for setup instructions.
from cuga import CugaAgent
from cuga.backend.cuga_graph.nodes.cuga_lite.combined_tool_provider import CombinedToolProvider
# Load from both tracker and registry (all available apps)
provider = CombinedToolProvider()
await provider.initialize() # Initialize connection to registry
agent = CugaAgent(tool_provider=provider)
# Load specific apps only from registry
provider = CombinedToolProvider(app_names=["slack", "github", "jira"])
await provider.initialize()
agent = CugaAgent(tool_provider=provider)Constructor Parameters:
Prop
Type
Complete Example
import asyncio
from cuga import CugaAgent
from cuga.backend.cuga_graph.nodes.cuga_lite.combined_tool_provider import CombinedToolProvider
async def main():
# Create provider that will execute tools against registry
provider = CombinedToolProvider(app_names=["slack", "github"])
# Initialize connection to registry
await provider.initialize()
# Create agent with the provider
agent = CugaAgent(tool_provider=provider)
# Agent can now use tools from both runtime tracker and registry
# Tools from registry will be executed through the registry's execution layer
result = await agent.invoke(
"Send a message to #general channel in Slack and create a GitHub issue"
)
print(result)
if __name__ == "__main__":
asyncio.run(main())Registry Execution Benefits
- Centralized Tool Management: All tools are managed through the registry
- Automatic Authentication: Registry handles API keys and authentication tokens
- Service Discovery: Automatically discovers available tools from registered services
- Unified Interface: Single interface for tools from different sources (OpenAPI, MCP)
- Runtime Flexibility: Can still use runtime tools when needed
Custom Tool Provider
You can create a custom tool provider by implementing the ToolProviderInterface to integrate with your own tool sources or add custom logic.
Implementation Example
from typing import List
from cuga.backend.cuga_graph.nodes.cuga_lite.tool_provider_interface import (
ToolProviderInterface,
AppDefinition
)
from langchain_core.tools import StructuredTool
from langchain_core.tools import tool
class MyCustomToolProvider(ToolProviderInterface):
def __init__(self, custom_tools: List[StructuredTool] = None):
self.custom_tools = custom_tools or []
self.apps = []
async def initialize(self):
# Initialize your tool source
# Connect to databases, APIs, or other services
# Validate tools and prepare app definitions
self.apps = [
AppDefinition(
name="custom_app",
url="https://api.example.com",
description="My custom application",
type="api"
)
]
async def get_apps(self) -> List[AppDefinition]:
# Return list of available apps from your custom source
return self.apps
async def get_tools(self, app_name: str) -> List[StructuredTool]:
# Return tools for the specified app
if app_name == "custom_app":
return self.custom_tools
return []
async def get_all_tools(self) -> List[StructuredTool]:
# Return all tools from all apps
return self.custom_tools
# Usage
@tool
def custom_tool(query: str) -> str:
"""A custom tool"""
return f"Result: {query}"
provider = MyCustomToolProvider(custom_tools=[custom_tool])
await provider.initialize()
agent = CugaAgent(tool_provider=provider)Tool Call Tracking with @tracked_tool
When implementing custom tool providers, you can use the @tracked_tool decorator to enable automatic tool call tracking. This is especially useful for observability, debugging, and auditing.
Why Use @tracked_tool?
When you use built-in providers like CombinedToolProvider, tool call tracking is handled automatically. However, for custom tool providers, you need to explicitly enable tracking using the @tracked_tool decorator.
Basic Usage
from cuga import CugaAgent, tracked_tool
from langchain_core.tools import tool
# Simple usage - just add the decorator
@tracked_tool
def my_tool(param: str) -> str:
return f"Result: {param}"
# With optional app_name for grouping
@tracked_tool(app_name="my_service")
def another_tool(x: int) -> int:
return x * 2
# Works with async functions
@tracked_tool(app_name="user_api")
async def fetch_user(user_id: int) -> dict:
return {"id": user_id, "name": "John"}Combining with LangChain @tool
You can combine @tracked_tool with LangChain's @tool decorator. Note that @tool should be the outer decorator:
from cuga import tracked_tool
from langchain_core.tools import tool
@tool
@tracked_tool(app_name="calculator")
def multiply(a: int, b: int) -> int:
'''Multiply two numbers together'''
return a * bCustom Tool Provider with Tracking
Here's a complete example of a custom tool provider with tracked tools:
from typing import List
from cuga import CugaAgent, tracked_tool
from cuga.backend.cuga_graph.nodes.cuga_lite.tool_provider_interface import (
ToolProviderInterface,
AppDefinition
)
from langchain_core.tools import StructuredTool, tool
class TrackedToolProvider(ToolProviderInterface):
def __init__(self):
self.apps = []
self.tools = []
async def initialize(self):
self.apps = [
AppDefinition(
name="user_service",
description="User management service",
type="api"
)
]
# Create tracked tools
@tool
@tracked_tool(app_name="user_service")
async def get_user(user_id: int) -> dict:
'''Get a user by ID'''
return {"id": user_id, "name": "John Doe", "email": "john@example.com"}
@tool
@tracked_tool(app_name="user_service")
async def list_users(limit: int = 10) -> list:
'''List users'''
return [{"id": i, "name": f"User {i}"} for i in range(limit)]
self.tools = [get_user, list_users]
async def get_apps(self) -> List[AppDefinition]:
return self.apps
async def get_tools(self, app_name: str) -> List[StructuredTool]:
if app_name == "user_service":
return self.tools
return []
async def get_all_tools(self) -> List[StructuredTool]:
return self.tools
# Usage with tracking enabled
async def main():
provider = TrackedToolProvider()
await provider.initialize()
agent = CugaAgent(tool_provider=provider)
# Enable tracking to see tool calls
result = await agent.invoke(
"Get user with ID 123",
track_tool_calls=True
)
print(result.answer)
# View tracked tool calls
for call in result.tool_calls:
print(f"Tool: {call['name']}")
print(f"App: {call['app_name']}")
print(f"Arguments: {call['arguments']}")
print(f"Result: {call['result']}")
print(f"Duration: {call['duration_ms']}ms")What Gets Tracked
The @tracked_tool decorator automatically captures:
Prop
Type
Decorator Parameters
Prop
Type