KEY INSIGHT
Shared tool access enables agents to collaborate effectively, but requires careful access control and result validation to prevent cascading failures.
Multi-agent systems frequently require shared access to tools: databases, APIs, file systems, computation resources. Agents must not only have access to tools but must share them coherently—results from one agent's tool usage inform another agent's operations. Poor tool sharing creates redundant work, inconsistent state, and error propagation.
Tool registration provides a catalog agents query before invoking capabilities. Each tool declares its interface, access requirements, rate limits, and usage cost. Agents select tools based on declared capabilities rather than hardcoded integrations.
Access control prevents unauthorized tool usage. Tools may require authentication, enforce quotas, or limit access to specific agents. Registration systems include policy evaluation determining whether requesting agents receive access.
Result caching prevents redundant tool invocations. When multiple agents query the same database, cached results serve subsequent requests. Cache invalidation strategies determine how stale cached data becomes before requiring fresh invocation.
Tool composition chains tools together. One tool's output becomes another's input. Chain execution requires type compatibility checking, error propagation, and transaction boundaries defining what constitutes success.
Result validation ensures tool outputs meet expectations. Tools can return malformed data, incomplete results, or unexpected types. Validators check outputs before passing them to dependent agents, catching failures early.
```python
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from datetime import datetime, timedelta
from enum import Enum
import hashlib
import json
class ToolCategory(Enum):
DATA_ACCESS = "data_access"
COMPUTATION = "computation"
EXTERNAL_API = "external_api"
FILE_SYSTEM = "file_system"
@dataclass
class ToolSpec:
name: str
category: ToolCategory
description: str
input_schema: dict
output_schema: dict
required_permissions: list[str] = field(default_factory=list)
rate_limit_per_minute: int = 60
cost_per_invocation: float = 0.0
@dataclass
class ToolResult:
tool_name: str
success: bool
output: Any
cached: bool = False
execution_time_ms: float = 0.0
timestamp: datetime = field(default_factory=datetime.utcnow)
error: Optional[str] = None
class ToolRegistry:
def __init__(self):
self.tools: dict[str, ToolSpec] = {}
self.implementations: dict[str, Callable] = {}
self.cache: dict[str, ToolResult] = {}
self.cache_ttl = timedelta(minutes=5)
def register(self, spec: ToolSpec, implementation: Callable) -> None:
self.tools[spec.name] = spec
self.implementations[spec.name] = implementation
def get_tool_spec(self, name: str) -> Optional[ToolSpec]:
return self.tools.get(name)
def list_tools(self, category: Optional[ToolCategory] = None) -> list[ToolSpec]:
if category:
return [t for t in self.tools.values() if t.category == category]
return list(self.tools.values())
def find_tools(self, required_capabilities: list[str]) -> list[ToolSpec]:
"""Find tools matching capability requirements"""
matching = []
for tool in self.tools.values():
# Simple text matching - real version would use capability graphs
if all(cap.lower() in tool.description.lower()
for cap in required_capabilities):
matching.append(tool)
return matching
class ToolAccessController:
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.permissions: dict[str, set[str]] = {} # agent_id -> allowed_tools
self.usage_counts: dict[str, dict[str, int]] = {} # agent_id -> tool_name -> count
def grant_permission(self, agent_id: str, tool_names: list[str]) -> None:
if agent_id not in self.permissions:
self.permissions[agent_id] = set()
self.permissions[agent_id].update(tool_names)
def can_access(self, agent_id: str, tool_name: str) -> bool:
allowed = self.permissions.get(agent_id, set())
if "*" in allowed:
return True
return tool_name in allowed
def check_rate_limit(self, agent_id: str, tool_name: str) -> bool:
spec = self.registry.get_tool_spec(tool_name)
if not spec:
return False
current_count = self.usage_counts.get(agent_id, {}).get(tool_name, 0)
return current_count < spec.rate_limit_per_minute
def record_usage(self, agent_id: str, tool_name: str) -> None:
if agent_id not in self.usage_counts:
self.usage_counts[agent_id] = {}
self.usage_counts[agent_id][tool_name] = \
self.usage_counts[agent_id].get(tool_name, 0) + 1
class ToolExecutor:
def __init__(
self,
registry: ToolRegistry,
controller: ToolAccessController,
validator: Optional[Any] = None
):
self.registry = registry
self.controller = controller
self.validator = validator
async def execute(
self,
agent_id: str,
tool_name: str,
parameters: dict
) -> ToolResult:
start_time = datetime.utcnow()
# Access check
if not self.controller.can_access(agent_id, tool_name):
return ToolResult(
tool_name=tool_name,
success=False,
output=None,
error=f"Agent {agent_id} lacks permission for {tool_name}"
)
# Rate limit check
if not self.controller.check_rate_limit(agent_id, tool_name):
return ToolResult(
tool_name=tool_name,
success=False,
output=None,
error=f"Rate limit exceeded for {tool_name}"
)
# Generate cache key
cache_key = self._generate_cache_key(tool_name, parameters)
# Check cache
if cache_key in self.registry.cache:
cached_result = self.registry.cache[cache_key]
if datetime.utcnow() - cached_result.timestamp < self.registry.cache_ttl:
cached_result.cached = True
return cached_result
# Execute tool
spec = self.registry.get_tool_spec(tool_name)
implementation = self.registry.implementations.get(tool_name)
if not spec or not implementation:
return ToolResult(
tool_name=tool_name,
success=False,
output=None,
error=f"Tool {tool_name} not found"
)
try:
output = await implementation(parameters)
# Validate output if validator present
if self.validator:
validation_result = self.validator.validate(
tool_name,
output,
spec.output_schema
)
if not validation_result.valid:
return ToolResult(
tool_name=tool_name,
success=False,
output=output,
error=f"Validation failed: {validation_result.errors}"
)
result = ToolResult(
tool_name=tool_name,
success=True,
output=output,
execution_time_ms=(datetime.utcnow() - start_time).total_seconds() * 1000
)
except Exception as e:
result = ToolResult(
tool_name=tool_name,
success=False,
output=None,
error=str(e),
execution_time_ms=(datetime.utcnow() - start_time).total_seconds() * 1000
)
# Record usage
self.controller.record_usage(agent_id, tool_name)
# Cache result
self.registry.cache[cache_key] = result
return result
def _generate_cache_key(self, tool_name: str, parameters: dict) -> str:
"""Generate deterministic cache key for request"""
canonical = json.dumps(parameters, sort_keys=True)
return hashlib.sha256(f"{tool_name}:{canonical}".encode()).hexdigest()
# Example tool definitions and usage
registry = ToolRegistry()
controller = ToolAccessController(registry)
executor = ToolExecutor(registry, controller)
async def database_query_impl(params: dict) -> list[dict]:
# Implementation would query actual database
return [{"id": 1, "name": "example"}]
registry.register(
ToolSpec(
name="query_database",
category=ToolCategory.DATA_ACCESS,
description="Execute SQL queries against the product database",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string"},
"parameters": {"type": "array"}
},
"required": ["query"]
},
output_schema={
"type": "array",
"items": {"type": "object"}
}
),
database_query_impl
)
controller.grant_permission("data_agent", ["query_database"])
controller.grant_permission("analytics_agent", ["query_database"])
result = await executor.execute(
"data_agent",
"query_database",
{"query": "SELECT * FROM products WHERE active = ?", "parameters": [True]}
)
```