Coverage for stackone_ai/feedback/tool.py: 95%
79 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-24 09:48 +0000
1"""Feedback collection tool for StackOne."""
3# TODO: Remove when Python 3.9 support is dropped
4from __future__ import annotations
6import json
8from pydantic import BaseModel, Field, field_validator
10from ..models import (
11 ExecuteConfig,
12 JsonDict,
13 ParameterLocation,
14 StackOneError,
15 StackOneTool,
16 ToolParameters,
17)
20class FeedbackInput(BaseModel):
21 """Input schema for feedback tool."""
23 feedback: str = Field(..., min_length=1, description="User feedback text")
24 account_id: str | list[str] = Field(..., description="Account identifier(s) - single ID or list of IDs")
25 tool_names: list[str] = Field(..., min_length=1, description="List of tool names")
27 @field_validator("feedback")
28 @classmethod
29 def validate_feedback(cls, v: str) -> str:
30 """Validate that feedback is non-empty after trimming."""
31 trimmed = v.strip()
32 if not trimmed:
33 raise ValueError("Feedback must be a non-empty string")
34 return trimmed
36 @field_validator("account_id")
37 @classmethod
38 def validate_account_id(cls, v: str | list[str]) -> list[str]:
39 """Validate and normalize account ID(s) to a list."""
40 if isinstance(v, str):
41 trimmed = v.strip()
42 if not trimmed:
43 raise ValueError("Account ID must be a non-empty string")
44 return [trimmed]
46 if isinstance(v, list): 46 ↛ 54line 46 didn't jump to line 54 because the condition on line 46 was always true
47 if not v:
48 raise ValueError("At least one account ID is required")
49 cleaned = [str(item).strip() for item in v if str(item).strip()]
50 if not cleaned:
51 raise ValueError("At least one valid account ID is required")
52 return cleaned
54 raise ValueError("Account ID must be a string or list of strings")
56 @field_validator("tool_names")
57 @classmethod
58 def validate_tool_names(cls, v: list[str]) -> list[str]:
59 """Validate and clean tool names."""
60 cleaned = [name.strip() for name in v if name.strip()]
61 if not cleaned:
62 raise ValueError("At least one tool name is required")
63 return cleaned
66class FeedbackTool(StackOneTool):
67 """Extended tool for collecting feedback with enhanced validation."""
69 def execute(
70 self, arguments: str | JsonDict | None = None, *, options: JsonDict | None = None
71 ) -> JsonDict:
72 """
73 Execute the feedback tool with enhanced validation.
75 If multiple account IDs are provided, sends the same feedback to each account individually.
77 Args:
78 arguments: Tool arguments as string or dict
79 options: Execution options
81 Returns:
82 Combined response from all API calls
84 Raises:
85 StackOneError: If validation or API call fails
86 """
87 try:
88 # Parse input
89 if isinstance(arguments, str):
90 raw_params = json.loads(arguments)
91 else:
92 raw_params = arguments or {}
94 # Validate with Pydantic
95 parsed_params = FeedbackInput(**raw_params)
97 # Get list of account IDs (already normalized by validator)
98 account_ids = parsed_params.account_id
99 feedback = parsed_params.feedback
100 tool_names = parsed_params.tool_names
102 # If only one account ID, use the parent execute method
103 if len(account_ids) == 1:
104 validated_arguments = {
105 "feedback": feedback,
106 "account_id": account_ids[0],
107 "tool_names": tool_names,
108 }
109 return super().execute(validated_arguments, options=options)
111 # Multiple account IDs - send to each individually
112 results = []
113 errors = []
115 for account_id in account_ids:
116 try:
117 validated_arguments = {
118 "feedback": feedback,
119 "account_id": account_id,
120 "tool_names": tool_names,
121 }
122 result = super().execute(validated_arguments, options=options)
123 results.append({"account_id": account_id, "status": "success", "result": result})
124 except Exception as exc:
125 error_msg = str(exc)
126 errors.append({"account_id": account_id, "status": "error", "error": error_msg})
127 results.append({"account_id": account_id, "status": "error", "error": error_msg})
129 # Return combined results
130 return {
131 "message": f"Feedback sent to {len(account_ids)} account(s)",
132 "total_accounts": len(account_ids),
133 "successful": len([r for r in results if r["status"] == "success"]),
134 "failed": len(errors),
135 "results": results,
136 }
138 except json.JSONDecodeError as exc:
139 raise StackOneError(f"Invalid JSON in arguments: {exc}") from exc
140 except ValueError as exc:
141 raise StackOneError(f"Validation error: {exc}") from exc
142 except Exception as error:
143 if isinstance(error, StackOneError): 143 ↛ 145line 143 didn't jump to line 145 because the condition on line 143 was always true
144 raise
145 raise StackOneError(f"Error executing feedback tool: {error}") from error
148def create_feedback_tool(
149 api_key: str,
150 account_id: str | None = None,
151 base_url: str = "https://api.stackone.com",
152) -> FeedbackTool:
153 """
154 Create a feedback collection tool.
156 Args:
157 api_key: API key for authentication
158 account_id: Optional account ID
159 base_url: Base URL for the API
161 Returns:
162 FeedbackTool configured for feedback collection
163 """
164 name = "meta_collect_tool_feedback"
165 description = (
166 "Collects user feedback on StackOne tool performance. "
167 'First ask the user, "Are you ok with sending feedback to StackOne?" '
168 "and mention that the LLM will take care of sending it. "
169 "Call this tool only when the user explicitly answers yes."
170 )
172 parameters = ToolParameters(
173 type="object",
174 properties={
175 "account_id": {
176 "oneOf": [
177 {
178 "type": "string",
179 "description": 'Single account identifier (e.g., "acc_123456")',
180 },
181 {
182 "type": "array",
183 "items": {"type": "string"},
184 "description": "List of account identifiers for multiple accounts",
185 },
186 ],
187 "description": "Account identifier(s) - single ID or list of IDs",
188 },
189 "feedback": {
190 "type": "string",
191 "description": "Verbatim feedback from the user about their experience with StackOne tools.",
192 },
193 "tool_names": {
194 "type": "array",
195 "items": {
196 "type": "string",
197 },
198 "description": "Array of tool names being reviewed",
199 },
200 },
201 )
203 execute_config = ExecuteConfig(
204 name=name,
205 method="POST",
206 url=f"{base_url}/ai/tool-feedback",
207 body_type="json",
208 parameter_locations={
209 "feedback": ParameterLocation.BODY,
210 "account_id": ParameterLocation.BODY,
211 "tool_names": ParameterLocation.BODY,
212 },
213 )
215 # Create instance by calling parent class __init__ directly since FeedbackTool is a subclass
216 tool = FeedbackTool.__new__(FeedbackTool)
217 StackOneTool.__init__(
218 tool,
219 description=description,
220 parameters=parameters,
221 _execute_config=execute_config,
222 _api_key=api_key,
223 _account_id=account_id,
224 )
226 return tool