Tool Discovery
Dynamic tool discovery, registration, and capability matching for agents.
Discovery Mechanisms
1. Static Registration (Compile-Time)
# Explicit registration
registry.register(WebSearchTool())
registry.register(CodeExecutionTool())
registry.register(DatabaseQueryTool())
2. Entry Points (Plugin System)
# pyproject.toml
[project.entry-points."myapp.tools"]
web_search = "myapp.tools.web:WebSearchTool"
github = "myapp.tools.github:GitHubTool"
slack = "myapp.tools.slack:SlackTool"
custom = "myplugin.tools:CustomTool"
# Discovery
from importlib.metadata import entry_points
def load_plugins(registry: ToolRegistry, group: str = "myapp.tools"):
for ep in entry_points(group=group):
tool_class = ep.load()
registry.register(tool_class(), category=ep.name)
3. Module Scanning
def discover_tools(package_name: str) -> list[Tool]:
"""Scan package for Tool subclasses"""
tools = []
package = importlib.import_module(package_name)
for _, module_name, _ in pkgutil.iter_modules(package.__path__):
module = importlib.import_module(f"{package_name}.{module_name}")
for name in dir(module):
obj = getattr(module, name)
if (inspect.isclass(obj) and
issubclass(obj, Tool) and
obj is not Tool):
tools.append(obj())
return tools
3. Configuration-Driven
# tools.yaml
tools:
- name: web_search
class: myapp.tools.web.WebSearchTool
config:
provider: serper
api_key_env: SERPER_API_KEY
enabled: true
category: research
- name: code_exec
class: myapp.tools.code.CodeExecutionTool
config:
sandbox: e2b
timeout: 60
enabled: true
category: development
- name: github
class: myapp.tools.github.GitHubTool
config:
app_id_env: GITHUB_APP_ID
private_key_env: GITHUB_PRIVATE_KEY
enabled: false
category: development
class ConfigDrivenRegistry:
def __init__(self, config_path: str):
self.config = self._load_config(config_path)
self.registry = ToolRegistry()
def load(self):
for tool_config in self.config["tools"]:
if not tool_config.get("enabled", True):
continue
tool_class = self._import_class(tool_config["class"])
tool_instance = tool_class(**tool_config.get("config", {}))
self.registry.register(
tool_instance,
category=tool_config.get("category", "general")
)
Capability Matching
Semantic Matching
class SemanticToolMatcher:
def __init__(self, embedder: EmbeddingModel):
self.embedder = embedder
self.tool_embeddings: dict[str, np.ndarray] = {}
def index_tools(self, tools: list[Tool]):
for tool in tools:
# Embed: name + description
text = f"{tool.name}: {tool.description}"
self.tool_embeddings[tool.name] = self.embedder.embed(text)
def match(self, query: str, top_k: int = 5) -> list[tuple[Tool, float]]:
query_emb = self.embedder.embed(query)
scores = []
for name, emb in self.tool_embeddings.items():
score = cosine_similarity(query_emb, emb)
scores.append((name, score))
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:top_k]
Capability Tags
@dataclass
class ToolCapability:
name: str
tags: set[str] # e.g., {"read", "web", "safe", "fast"}
requires_auth: bool = False
cost_level: Literal["free", "low", "medium", "high"] = "free"
latency_class: Literal["instant", "fast", "medium", "slow"] = "fast"
class TaggedToolRegistry:
def __init__(self):
self.tools: dict[str, Tool] = {}
self.capabilities: dict[str, ToolCapability] = {}
def register(self, tool: Tool, capability: ToolCapability):
self.tools[tool.name] = tool
self.capabilities[tool.name] = capability
def find_by_capability(
self,
required_tags: set[str] = None,
forbidden_tags: set[str] = None,
max_cost: str = "high",
max_latency: str = "slow",
requires_auth: bool = None
) -> list[Tool]:
candidates = []
for name, tool in self.tools.items():
cap = self.capabilities[name]
# Check tags
if required_tags and not required_tags.issubset(cap.tags):
continue
if forbidden_tags and forbidden_tags.intersection(cap.tags):
continue
# Check cost
cost_order = {"free": 0, "low": 1, "medium": 2, "high": 3}
if cost_order[cap.cost_level] > cost_order[max_cost]:
continue
# Check latency
latency_order = {"instant": 0, "fast": 1, "medium": 2, "slow": 3}
if latency_order[cap.latency_class] > latency_order[max_latency]:
continue
# Check auth
if requires_auth is not None and cap.requires_auth != requires_auth:
continue
candidates.append(self.tools[name])
return candidates
Dynamic Tool Loading
Runtime Loading
class DynamicToolLoader:
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.loaded_modules: set[str] = set()
async def load_tool(self, module_path: str, class_name: str) -> Tool:
"""Load tool at runtime"""
module = importlib.import_module(module_path)
tool_class = getattr(module, class_name)
tool_instance = tool_class()
self.registry.register(tool_instance)
return tool_instance
async def load_from_url(self, url: str) -> Tool:
"""Load tool from remote source (with verification)"""
# Download, verify signature, load
code = await self._download_verified(url)
module = self._create_module(code)
tool_class = getattr(module, "Tool")
return tool_class()
def unload_tool(self, tool_name: str):
if tool_name in self.registry._tools:
tool = self.registry._tools.pop(tool_name)
if hasattr(tool, 'shutdown'):
asyncio.create_task(tool.shutdown())
Hot Reloading
class HotReloader:
def __init__(self, registry: ToolRegistry, watch_paths: list[str]):
self.registry = registry
self.watch_paths = watch_paths
self.file_times: dict[str, float] = {}
async def watch(self, interval: float = 5.0):
while True:
await asyncio.sleep(interval)
await self.check_changes()
async def check_changes(self):
for path in self.watch_paths:
for file_path in Path(path).rglob("*.py"):
mtime = file_path.stat().st_mtime
if file_path not in self.file_times:
self.file_times[file_path] = mtime
elif mtime > self.file_times[file_path]:
self.file_times[file_path] = mtime
await self.reload_module(file_path)
async def reload_module(self, file_path: Path):
# Reload module and re-register tools
module_name = self._path_to_module(file_path)
importlib.reload(sys.modules[module_name])
# Re-discover tools in module
await self._rediscover_module(module_name)
Capability Negotiation
Agent-Tool Negotiation
class CapabilityNegotiator:
def __init__(self, registry: ToolRegistry):
self.registry = registry
def negotiate(
self,
agent_requirements: AgentRequirements,
available_tools: list[Tool]
) -> NegotiationResult:
"""
Match agent requirements to available tools.
"""
matched = []
missing = []
alternatives = {}
for req in agent_requirements.required_capabilities:
matches = self._find_matches(req, available_tools)
if matches:
matched.append((req, matches[0]))
else:
missing.append(req)
# Find alternatives
alt = self._find_alternatives(req, available_tools)
if alt:
alternatives[req.name] = alt
return NegotiationResult(
matched=matched,
missing=missing,
alternatives=alternatives,
success=len(missing) == 0
)
def _find_matches(self, req: CapabilityRequirement, tools: list[Tool]) -> list[Tool]:
return [
t for t in tools
if self._matches_requirement(t, req)
]
Capability Requirements
@dataclass
class CapabilityRequirement:
name: str
required_tags: set[str] = field(default_factory=set)
forbidden_tags: set[str] = field(default_factory=set)
max_cost: str = "high"
max_latency: str = "slow"
requires_auth: bool | None = None
minimum_version: str | None = None
@dataclass
class AgentRequirements:
agent_name: str
required_capabilities: list[CapabilityRequirement]
preferred_tools: list[str] = field(default_factory=list)
forbidden_tools: list[str] = field(default_factory=list)
Marketplace / Registry
Central Registry
class ToolMarketplace:
def __init__(self):
self.tools: dict[str, MarketplaceTool] = {}
self.categories: dict[str, list[str]] = defaultdict(list)
self.ratings: dict[str, list[Rating]] = defaultdict(list)
def publish(self, tool: MarketplaceTool):
self.tools[tool.name] = tool
self.categories[tool.category].append(tool.name)
def search(self, query: str, category: str = None, tags: list[str] = None) -> list[MarketplaceTool]:
candidates = self.tools.values()
if category:
candidates = [t for t in candidates if t.category == category]
if tags:
candidates = [t for t in candidates if set(tags).issubset(t.tags)]
# Rank by relevance (simplified)
return sorted(candidates, key=lambda t: t.popularity, reverse=True)
def install(self, tool_name: str, target_registry: ToolRegistry) -> Tool:
tool = self.tools[tool_name]
# Download, verify, instantiate
tool_instance = tool.instantiate()
target_registry.register(tool_instance)
return tool_instance
@dataclass
class MarketplaceTool:
name: str
version: str
description: str
category: str
tags: list[str]
author: str
license: str
source_url: str
checksum: str
popularity: int = 0
ratings: list[Rating] = field(default_factory=list)
def instantiate(self) -> Tool:
# Download, verify checksum, instantiate
pass
Private Registry
class PrivateRegistry:
def __init__(self, storage: StorageBackend, auth: AuthProvider):
self.storage = storage
self.auth = auth
async def push(self, tool: Tool, version: str, tenant: str):
# Package tool
package = self._package_tool(tool, version)
# Store with tenant isolation
await self.storage.put(f"tenants/{tenant}/tools/{tool.name}/{version}", package)
async def pull(self, tool_name: str, version: str, tenant: str) -> Tool:
package = await self.storage.get(f"tenants/{tenant}/tools/{tool_name}/{version}")
return self._unpack_tool(package)
async def list_versions(self, tool_name: str, tenant: str) -> list[str]:
return await self.storage.list(f"tenants/{tenant}/tools/{tool_name}/")
Discovery UI/CLI
CLI
# List available tools
myapp tools list
# Search
myapp tools search "web search" --category research
# Install
myapp tools install web_search_v2
# Update
myapp tools update web_search
# Info
myapp tools info web_search
Web UI
# FastAPI endpoints
@app.get("/tools")
async def list_tools(category: str = None, tags: list[str] = Query(None)):
return registry.search(category=category, tags=tags)
@app.get("/tools/{tool_name}")
async def get_tool(tool_name: str):
return registry.get(tool_name)
@app.post("/tools/{tool_name}/install")
async def install_tool(tool_name: str, current_user: User):
tool = marketplace.install(tool_name, user_registry)
return {"status": "installed", "tool": tool.name}
@app.get("/tools/{tool_name}/versions")
async def list_versions(tool_name: str):
return registry.list_versions(tool_name)
Security
Verification
class ToolVerifier:
def __init__(self, trusted_keys: list[PublicKey]):
self.trusted_keys = trusted_keys
def verify(self, tool_package: ToolPackage) -> VerificationResult:
# 1. Verify signature
for key in self.trusted_keys:
if key.verify(tool_package.signature, tool_package.payload):
break
else:
return VerificationResult(valid=False, reason="Invalid signature")
# 2. Verify checksum
if not self._verify_checksum(tool_package):
return VerificationResult(valid=False, reason="Checksum mismatch")
# 3. Scan for malware/suspicious patterns
scan_result = self._security_scan(tool_package.code)
if not scan_result.clean:
return VerificationResult(valid=False, reason=f"Security scan failed: {scan_result.issues}")
return VerificationResult(valid=True)
def _security_scan(self, code: str) -> ScanResult:
# Check for suspicious patterns
suspicious = [
"eval(", "exec(", "__import__", "subprocess",
"os.system", "open(", "__builtins__"
]
issues = [s for s in suspicious if s in code]
return ScanResult(clean=len(issues) == 0, issues=issues)
Sandbox Verification
async def verify_in_sandbox(tool: Tool, test_cases: list[dict]) -> VerificationResult:
"""Run tool in isolated sandbox with test cases"""
sandbox = create_sandbox()
for test_case in test_cases:
try:
result = await sandbox.run_tool(tool, test_case["input"])
expected = test_case.get("expected")
if expected and not matches(result, expected):
return VerificationResult(valid=False, reason=f"Output mismatch: {result} != {expected}")
except Exception as e:
return VerificationResult(valid=False, reason=f"Execution failed: {e}")
return VerificationResult(valid=True)
Related Notes
- Custom Tools — Building tools
- Function Calling — Tools as functions
- MCP Protocol — MCP discovery
- LangGraph — Tool discovery in LangGraph
- AgentKit — AgentKit tool discovery
- OpenAI Agents SDK — OpenAI SDK discovery
- PydanticAI — PydanticAI tool discovery
- Control Flow — Tool selection in control flow
- Security — Tool security
- Observability — Discovery observability
- Research Directions — Autonomous tool discovery
Sources: Plugin architectures, entry points, semantic search, capability-based routing, plugin marketplaces (VS Code, Chrome, WordPress), security scanning (Bandit, Semgrep), sandbox verification, dynamic loading patterns.