# Tool Discovery

Dynamic tool discovery, registration, and capability matching for agents.

---

## Discovery Mechanisms

### 1. Static Registration (Compile-Time)
```python
# Explicit registration
registry.register(WebSearchTool())
registry.register(CodeExecutionTool())
registry.register(DatabaseQueryTool())
```

### 2. Entry Points (Plugin System)
```python
# 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"
```

```python
# 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
```python
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
```yaml
# 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
```

```python
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
```python
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
```python
@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
```python
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
```python
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
```python
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
```python
@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
```python
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
```python
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
```bash
# 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
```python
# 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
```python
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
```python
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

- [[ToolUse/Custom-Tools.md]] — Building tools
- [[ToolUse/Function-Calling.md]] — Tools as functions
- [[ToolUse/MCP-Protocol.md]] — MCP discovery
- [[Frameworks/LangGraph.md]] — Tool discovery in LangGraph
- [[Frameworks/AgentKit.md]] — AgentKit tool discovery
- [[Frameworks/OpenAI-Agents-SDK.md]] — OpenAI SDK discovery
- [[Frameworks/PydanticAI.md]] — PydanticAI tool discovery
- [[Architectures/Control-Flow.md]] — Tool selection in control flow
- [[Production/Security.md]] — Tool security
- [[Production/Observability.md]] — Discovery observability
- [[Frontiers/Research-Directions.md]] — 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.*