Advanced Tool Patterns

1from dataclasses import dataclass
2from typing import Any
3
4class CompositeToolBuilder:
5    """Build composite tools from multiple sub-tools."""
6
7    def __init__(self, registry: ToolRegistry):
8        self.registry = registry
9
10    def create_composite(
11        self,
12        name: str,
13        description: str,
14        steps: list[dict],
15        parameters: dict
16    ) -> Tool:
17        """Create a composite tool from a sequence of steps."""
18
19        async def execute(**kwargs):
20            context = {"input": kwargs, "results": {}}
21
22            for step in steps:
23                tool_name = step["tool"]
24                param_mapping = step.get("params", {})
25
26                # Build parameters from context
27                params = {}
28                for param, source in param_mapping.items():
29                    params[param] = self._resolve_value(source, context)
30
31                # Execute step
32                result = await self.registry.get(tool_name).function(**params)
33                context["results"][step.get("name", tool_name)] = result
34
35            # Return final result
36            return_expr = steps[-1].get("return", "results")
37            return self._resolve_value(return_expr, context)
38
39        return Tool(
40            name=name,
41            description=description,
42            parameters=parameters,
43            function=execute
44        )
45
46    def _resolve_value(self, expr: str, context: dict) -> Any:
47        """Resolve value from context using dot notation."""
48        if not expr.startswith("$"):
49            return expr
50
51        path = expr[1:].split(".")
52        value = context
53
54        for key in path:
55            if isinstance(value, dict):
56                value = value.get(key)
57            else:
58                value = getattr(value, key, None)
59
60        return value
61
62
63# Example: Create a "deploy" composite tool
64builder = CompositeToolBuilder(registry)
65
66deploy_tool = builder.create_composite(
67    name="deploy_to_production",
68    description="""
69Deploy code to production environment.
70
71This tool:
721. Runs the test suite
732. Builds the application
743. Deploys to production
754. Verifies deployment
76
77Only use after changes have been reviewed.
78""",
79    parameters={
80        "type": "object",
81        "properties": {
82            "branch": {"type": "string", "default": "main"},
83            "skip_tests": {"type": "boolean", "default": False}
84        }
85    },
86    steps=[
87        {
88            "name": "tests",
89            "tool": "run_tests",
90            "params": {"path": "./", "coverage": True},
91            "skip_if": "$input.skip_tests"
92        },
93        {
94            "name": "build",
95            "tool": "build_app",
96            "params": {"environment": "production"}
97        },
98        {
99            "name": "deploy",
100            "tool": "deploy_to_server",
101            "params": {
102                "artifact": "$results.build.artifact_path",
103                "server": "production"
104            }
105        },
106        {
107            "name": "verify",
108            "tool": "health_check",
109            "params": {"url": "https://api.example.com/health"},
110            "return": "$results.verify"
111        }
112    ]
113)

1class TransactionalComposite:
2    """Composite tool with rollback on failure."""
3
4    def __init__(self, registry: ToolRegistry):
5        self.registry = registry
6
7    async def execute_with_rollback(
8        self,
9        steps: list[dict],
10        context: dict
11    ) -> dict:
12        """Execute steps with automatic rollback on failure."""
13
14        completed = []
15
16        try:
17            for step in steps:
18                result = await self._execute_step(step, context)
19                completed.append({
20                    "step": step,
21                    "result": result
22                })
23                context["results"][step["name"]] = result
24
25            return {"success": True, "results": context["results"]}
26
27        except Exception as e:
28            # Rollback in reverse order
29            for completed_step in reversed(completed):
30                rollback = completed_step["step"].get("rollback")
31                if rollback:
32                    try:
33                        await self._execute_step(rollback, context)
34                    except Exception:
35                        pass  # Best effort rollback
36
37            return {
38                "success": False,
39                "error": str(e),
40                "rolled_back": [s["step"]["name"] for s in completed]
41            }
42
43
44# Example: Atomic database migration
45migration_steps = [
46    {
47        "name": "backup",
48        "tool": "create_backup",
49        "params": {"database": "$input.db_name"}
50    },
51    {
52        "name": "migrate",
53        "tool": "run_migration",
54        "params": {"script": "$input.migration_file"},
55        "rollback": {
56            "tool": "restore_backup",
57            "params": {"backup_id": "$results.backup.id"}
58        }
59    },
60    {
61        "name": "verify",
62        "tool": "verify_schema",
63        "params": {"expected": "$input.expected_schema"}
64    }
65]

1from typing import Any, Callable
2import inspect
3
4class DynamicToolFactory:
5    """Generate tools dynamically based on configuration."""
6
7    def __init__(self, registry: ToolRegistry):
8        self.registry = registry
9
10    def from_openapi_spec(self, spec: dict) -> list[Tool]:
11        """Generate tools from OpenAPI specification."""
12
13        tools = []
14
15        for path, methods in spec.get("paths", {}).items():
16            for method, details in methods.items():
17                if method not in ["get", "post", "put", "delete"]:
18                    continue
19
20                tool = self._create_api_tool(
21                    path=path,
22                    method=method,
23                    details=details,
24                    base_url=spec.get("servers", [{}])[0].get("url", "")
25                )
26                tools.append(tool)
27                self.registry._tools[tool.name] = tool
28
29        return tools
30
31    def _create_api_tool(
32        self,
33        path: str,
34        method: str,
35        details: dict,
36        base_url: str
37    ) -> Tool:
38        """Create a tool from an API endpoint."""
39
40        operation_id = details.get("operationId", f"{method}_{path}")
41        name = self._to_snake_case(operation_id)
42
43        # Build parameters schema from spec
44        parameters = self._build_parameters_schema(details)
45
46        async def execute(**kwargs):
47            import aiohttp
48
49            url = base_url + path
50            # Substitute path parameters
51            for key, value in kwargs.items():
52                url = url.replace(f"{{{key}}}", str(value))
53
54            async with aiohttp.ClientSession() as session:
55                request_method = getattr(session, method)
56                async with request_method(url, json=kwargs) as response:
57                    return await response.json()
58
59        return Tool(
60            name=name,
61            description=details.get("summary", details.get("description", "")),
62            parameters=parameters,
63            function=execute
64        )
65
66    def from_database_schema(self, schema: dict) -> list[Tool]:
67        """Generate CRUD tools from database schema."""
68
69        tools = []
70
71        for table_name, columns in schema.items():
72            # Create tool
73            tools.append(self._create_crud_tool(table_name, columns, "create"))
74            # Read tool
75            tools.append(self._create_crud_tool(table_name, columns, "read"))
76            # Update tool
77            tools.append(self._create_crud_tool(table_name, columns, "update"))
78            # Delete tool
79            tools.append(self._create_crud_tool(table_name, columns, "delete"))
80            # List tool
81            tools.append(self._create_crud_tool(table_name, columns, "list"))
82
83        return tools
84
85    def _create_crud_tool(
86        self,
87        table: str,
88        columns: dict,
89        operation: str
90    ) -> Tool:
91        """Create a CRUD tool for a database table."""
92
93        name = f"{operation}_{table}"
94
95        if operation == "create":
96            description = f"Create a new {table} record"
97            params = self._columns_to_schema(columns, required=True)
98        elif operation == "read":
99            description = f"Get a {table} record by ID"
100            params = {"type": "object", "properties": {"id": {"type": "string"}}, "required": ["id"]}
101        elif operation == "update":
102            description = f"Update a {table} record"
103            params = self._columns_to_schema(columns, required=False)
104            params["properties"]["id"] = {"type": "string"}
105            params["required"] = ["id"]
106        elif operation == "delete":
107            description = f"Delete a {table} record"
108            params = {"type": "object", "properties": {"id": {"type": "string"}}, "required": ["id"]}
109        else:  # list
110            description = f"List {table} records with optional filters"
111            params = {
112                "type": "object",
113                "properties": {
114                    "limit": {"type": "integer", "default": 10},
115                    "offset": {"type": "integer", "default": 0},
116                    "filters": {"type": "object"}
117                }
118            }
119
120        async def execute(**kwargs):
121            # Implementation would use actual database
122            return {"operation": operation, "table": table, "params": kwargs}
123
124        return Tool(name=name, description=description, parameters=params, function=execute)

1from dataclasses import dataclass
2from typing import Any, Callable
3
4@dataclass
5class PipelineStep:
6    """A step in a tool pipeline."""
7    tool_name: str
8    param_transform: Callable[[Any], dict] | None = None
9    result_transform: Callable[[Any], Any] | None = None
10    condition: Callable[[Any], bool] | None = None
11
12class ToolPipeline:
13    """Chain tools together in a pipeline."""
14
15    def __init__(self, registry: ToolRegistry):
16        self.registry = registry
17        self.steps: list[PipelineStep] = []
18
19    def add_step(
20        self,
21        tool_name: str,
22        param_transform: Callable[[Any], dict] | None = None,
23        result_transform: Callable[[Any], Any] | None = None,
24        condition: Callable[[Any], bool] | None = None
25    ) -> "ToolPipeline":
26        """Add a step to the pipeline (fluent interface)."""
27
28        self.steps.append(PipelineStep(
29            tool_name=tool_name,
30            param_transform=param_transform,
31            result_transform=result_transform,
32            condition=condition
33        ))
34        return self
35
36    async def execute(self, initial_input: Any) -> Any:
37        """Execute the pipeline."""
38
39        current_value = initial_input
40
41        for step in self.steps:
42            # Check condition
43            if step.condition and not step.condition(current_value):
44                continue
45
46            # Transform parameters
47            if step.param_transform:
48                params = step.param_transform(current_value)
49            elif isinstance(current_value, dict):
50                params = current_value
51            else:
52                params = {"input": current_value}
53
54            # Execute tool
55            tool = self.registry.get(step.tool_name)
56            result = await tool.function(**params)
57
58            # Transform result
59            if step.result_transform:
60                current_value = step.result_transform(result)
61            else:
62                current_value = result
63
64        return current_value
65
66
67# Example: Data processing pipeline
68pipeline = (
69    ToolPipeline(registry)
70    .add_step(
71        "fetch_data",
72        param_transform=lambda x: {"url": x["source_url"]}
73    )
74    .add_step(
75        "parse_csv",
76        param_transform=lambda x: {"content": x},
77        result_transform=lambda x: x["rows"]
78    )
79    .add_step(
80        "filter_data",
81        param_transform=lambda rows: {"data": rows, "condition": "status == 'active'"}
82    )
83    .add_step(
84        "aggregate",
85        param_transform=lambda rows: {"data": rows, "group_by": "category", "agg": "sum"}
86    )
87    .add_step(
88        "save_results",
89        param_transform=lambda x: {"data": x, "path": "/tmp/results.json"}
90    )
91)
92
93result = await pipeline.execute({"source_url": "https://example.com/data.csv"})

1class ParallelPipeline:
2    """Execute multiple pipelines in parallel and merge results."""
3
4    def __init__(self, registry: ToolRegistry):
5        self.registry = registry
6        self.branches: list[ToolPipeline] = []
7
8    def add_branch(self, pipeline: ToolPipeline) -> "ParallelPipeline":
9        """Add a parallel branch."""
10        self.branches.append(pipeline)
11        return self
12
13    async def execute(
14        self,
15        initial_input: Any,
16        merge_fn: Callable[[list[Any]], Any] | None = None
17    ) -> Any:
18        """Execute all branches in parallel."""
19
20        tasks = [branch.execute(initial_input) for branch in self.branches]
21        results = await asyncio.gather(*tasks, return_exceptions=True)
22
23        # Filter out exceptions
24        valid_results = [r for r in results if not isinstance(r, Exception)]
25
26        if merge_fn:
27            return merge_fn(valid_results)
28
29        return valid_results
30
31
32# Example: Gather data from multiple sources
33parallel = (
34    ParallelPipeline(registry)
35    .add_branch(
36        ToolPipeline(registry)
37        .add_step("search_google", param_transform=lambda x: {"query": x})
38    )
39    .add_branch(
40        ToolPipeline(registry)
41        .add_step("search_bing", param_transform=lambda x: {"query": x})
42    )
43    .add_branch(
44        ToolPipeline(registry)
45        .add_step("search_database", param_transform=lambda x: {"query": x})
46    )
47)
48
49results = await parallel.execute(
50    "Python best practices",
51    merge_fn=lambda results: list(set().union(*results))  # Deduplicate
52)

1from dataclasses import dataclass, field
2from typing import Any, Dict
3import uuid
4
5@dataclass
6class ToolSession:
7    """Stateful session for tools."""
8    id: str = field(default_factory=lambda: str(uuid.uuid4()))
9    state: dict = field(default_factory=dict)
10    created_at: datetime = field(default_factory=datetime.now)
11    last_accessed: datetime = field(default_factory=datetime.now)
12
13class StatefulToolManager:
14    """Manage stateful tool sessions."""
15
16    def __init__(self):
17        self.sessions: dict[str, ToolSession] = {}
18        self.tool_states: dict[str, dict[str, Any]] = {}
19
20    def create_session(self) -> str:
21        """Create a new tool session."""
22        session = ToolSession()
23        self.sessions[session.id] = session
24        return session.id
25
26    def get_state(self, session_id: str, tool_name: str) -> dict:
27        """Get tool state for a session."""
28        key = f"{session_id}:{tool_name}"
29        return self.tool_states.get(key, {})
30
31    def set_state(self, session_id: str, tool_name: str, state: dict):
32        """Set tool state for a session."""
33        key = f"{session_id}:{tool_name}"
34        self.tool_states[key] = state
35
36    def update_state(self, session_id: str, tool_name: str, updates: dict):
37        """Update tool state."""
38        current = self.get_state(session_id, tool_name)
39        current.update(updates)
40        self.set_state(session_id, tool_name, current)
41
42
43def stateful_tool(name: str, description: str):
44    """Decorator to create a stateful tool."""
45
46    def decorator(func):
47        async def wrapper(
48            session_id: str,
49            state_manager: StatefulToolManager,
50            **kwargs
51        ):
52            # Get current state
53            state = state_manager.get_state(session_id, name)
54
55            # Execute with state
56            result, new_state = await func(state=state, **kwargs)
57
58            # Save new state
59            if new_state:
60                state_manager.update_state(session_id, name, new_state)
61
62            return result
63
64        wrapper.__name__ = name
65        wrapper.__doc__ = description
66        return wrapper
67
68    return decorator
69
70
71# Example: Stateful code editor
72@stateful_tool(
73    name="code_editor",
74    description="A stateful code editor that remembers open files and changes"
75)
76async def code_editor(
77    state: dict,
78    action: str,
79    file_path: str | None = None,
80    content: str | None = None,
81    position: int | None = None
82) -> tuple[Any, dict]:
83    """
84    Stateful code editor.
85
86    Actions: open, read, write, close, list_open
87    State tracks: open_files, current_file, undo_history
88    """
89
90    open_files = state.get("open_files", {})
91    current_file = state.get("current_file")
92    undo_history = state.get("undo_history", [])
93
94    if action == "open":
95        # Read file and add to open files
96        with open(file_path) as f:
97            content = f.read()
98        open_files[file_path] = content
99        current_file = file_path
100        result = f"Opened {file_path}"
101
102    elif action == "read":
103        if file_path in open_files:
104            result = open_files[file_path]
105        else:
106            result = "File not open"
107
108    elif action == "write":
109        if current_file:
110            # Save for undo
111            undo_history.append({
112                "file": current_file,
113                "content": open_files[current_file]
114            })
115            open_files[current_file] = content
116            result = "Content updated"
117        else:
118            result = "No file selected"
119
120    elif action == "close":
121        if file_path in open_files:
122            del open_files[file_path]
123            if current_file == file_path:
124                current_file = None
125            result = f"Closed {file_path}"
126        else:
127            result = "File not open"
128
129    elif action == "list_open":
130        result = list(open_files.keys())
131
132    else:
133        result = f"Unknown action: {action}"
134
135    new_state = {
136        "open_files": open_files,
137        "current_file": current_file,
138        "undo_history": undo_history[-10:]  # Keep last 10
139    }
140
141    return result, new_state

1from dataclasses import dataclass
2import numpy as np
3
4@dataclass
5class ToolMatch:
6    """A potential tool match for a task."""
7    tool: Tool
8    relevance_score: float
9    reason: str
10
11class ToolDiscovery:
12    """Help agents find relevant tools."""
13
14    def __init__(self, registry: ToolRegistry, embedder):
15        self.registry = registry
16        self.embedder = embedder
17        self.tool_embeddings: dict[str, np.ndarray] = {}
18        self._build_embeddings()
19
20    def _build_embeddings(self):
21        """Pre-compute embeddings for all tools."""
22        for tool in self.registry.list_tools():
23            text = f"{tool.name}: {tool.description}"
24            self.tool_embeddings[tool.name] = self.embedder.embed(text)
25
26    def find_relevant_tools(
27        self,
28        query: str,
29        top_k: int = 5,
30        min_score: float = 0.5
31    ) -> list[ToolMatch]:
32        """Find tools relevant to a query."""
33
34        query_embedding = self.embedder.embed(query)
35        matches = []
36
37        for tool_name, tool_embedding in self.tool_embeddings.items():
38            score = self._cosine_similarity(query_embedding, tool_embedding)
39
40            if score >= min_score:
41                tool = self.registry.get(tool_name)
42                matches.append(ToolMatch(
43                    tool=tool,
44                    relevance_score=score,
45                    reason=self._explain_match(query, tool)
46                ))
47
48        matches.sort(key=lambda m: m.relevance_score, reverse=True)
49        return matches[:top_k]
50
51    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
52        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
53
54    def _explain_match(self, query: str, tool: Tool) -> str:
55        """Generate explanation for why tool matches."""
56        return f"Tool '{tool.name}' may help with: {tool.description[:100]}..."
57
58
59class ToolSelector:
60    """LLM-based tool selection for complex tasks."""
61
62    def __init__(self, llm, registry: ToolRegistry):
63        self.llm = llm
64        self.registry = registry
65
66    async def select_tools(
67        self,
68        task: str,
69        available_tools: list[Tool] | None = None
70    ) -> list[str]:
71        """Use LLM to select appropriate tools for a task."""
72
73        tools = available_tools or self.registry.list_tools()
74
75        tool_descriptions = "\n".join([
76            f"- {t.name}: {t.description[:200]}"
77            for t in tools
78        ])
79
80        prompt = f"""Given the following task and available tools, select which tools
81would be needed to complete the task. Return a JSON array of tool names.
82
83TASK: {task}
84
85AVAILABLE TOOLS:
86{tool_descriptions}
87
88Return ONLY a JSON array of tool names, e.g., ["tool1", "tool2"]
89"""
90
91        response = await self.llm.generate(prompt)
92
93        try:
94            import json
95            return json.loads(response)
96        except json.JSONDecodeError:
97            return []
98
99
100class AdaptiveToolProvider:
101    """Provide different tools based on context."""
102
103    def __init__(self, registry: ToolRegistry):
104        self.registry = registry
105        self.context_rules: list[tuple[Callable, list[str]]] = []
106
107    def add_rule(
108        self,
109        condition: Callable[[dict], bool],
110        tool_names: list[str]
111    ):
112        """Add a rule for tool filtering."""
113        self.context_rules.append((condition, tool_names))
114
115    def get_tools_for_context(self, context: dict) -> list[Tool]:
116        """Get tools applicable to current context."""
117
118        applicable_tools = set()
119
120        for condition, tool_names in self.context_rules:
121            if condition(context):
122                applicable_tools.update(tool_names)
123
124        if not applicable_tools:
125            # Default: all tools
126            return self.registry.list_tools()
127
128        return [
129            self.registry.get(name)
130            for name in applicable_tools
131            if self.registry.get(name)
132        ]
133
134
135# Example usage
136provider = AdaptiveToolProvider(registry)
137
138# Only provide file tools when working with code
139provider.add_rule(
140    condition=lambda ctx: ctx.get("mode") == "coding",
141    tool_names=["read_file", "write_file", "search_code", "execute_python"]
142)
143
144# Provide research tools for research mode
145provider.add_rule(
146    condition=lambda ctx: ctx.get("mode") == "research",
147    tool_names=["search_web", "fetch_url", "summarize"]
148)
149
150# Get tools based on current context
151context = {"mode": "coding", "language": "python"}
152tools = provider.get_tools_for_context(context)