Coverage for src/integry/resources/functions/api.py: 14%
81 statements
« prev ^ index » next coverage.py v7.6.1, created at 2025-03-26 01:26 +0500
1from typing import Any, Literal, Optional, List
3import httpx
5from integry.exceptions import FunctionCallError
6from integry.resources.base import BaseResource, AsyncPaginator
7from .types import (
8 Function,
9 FunctionCallOutput,
10 FunctionsPage,
11 IncludeOptions,
12 FunctionType,
13 PaginatedFunctionCallOutput,
14)
15from urllib.parse import urlencode
18class Functions(BaseResource):
20 def list(
21 self,
22 user_id: str,
23 app: Optional[str] = None,
24 connected_only: Optional[bool] = False,
25 type: Optional[Literal["ACTION", "QUERY"]] = None,
26 cursor: str = "",
27 include: Optional[IncludeOptions] = None,
28 ) -> AsyncPaginator[Function, FunctionsPage]:
29 """
30 Lists all functions.
32 Args:
33 user_id: The ID of the user.
34 app: The name of the app to filter the functions by.
35 connected_only: Whether to consider only the functions of the user's connected apps.
36 type: The type to filter functions by.
37 cursor: Provide the cursor from last page to fetch the next page of functions.
38 include: The fields to include with the functions.
40 Returns:
41 List of functions.
42 """
43 query_string = self._get_query_string(include, app, connected_only, type)
44 return AsyncPaginator(
45 self,
46 user_id,
47 query_string,
48 explicit_cursor=cursor,
49 model=Function,
50 paginated_response_model=FunctionsPage,
51 )
53 async def predict(
54 self,
55 prompt: str,
56 user_id: str,
57 variables: Optional[dict[str, Any]] = None,
58 predict_arguments: bool = False,
59 connected_only: Optional[bool] = False,
60 include: Optional[IncludeOptions] = None,
61 ) -> List[Function]:
62 """
63 Predicts a function based on the given prompt.
65 Args:
66 prompt: The prompt to use for predicting the function.
67 user_id: The ID of the user.
68 variables: The variables to use for auto-mapping the arguments.
69 Omit if you don't want to auto-map the arguments.
70 predict_arguments: Whether to predict the function's arguments.
71 connected_only: Whether to consider only the functions of the user's connected apps.
72 include: The fields to include with the function.
74 Returns:
75 A list containing the predicted function, or an empty list if no function was predicted.
76 """
77 query_string = self._get_query_string(
78 include, None, connected_only, None, predict_arguments
79 )
81 body: dict[str, Any] = {"prompt": prompt}
82 if variables:
83 body["_variables"] = variables
85 response = await self.http_client.post(
86 f"{self.name}/predict/{query_string}",
87 headers=self._get_signed_request_headers(user_id),
88 json=body,
89 )
90 data = self._get_response_data_or_raise(response)
92 predicted_functions = data.get("functions", [])
93 return [
94 Function(**function, _resource=self) for function in predicted_functions
95 ]
97 async def get(
98 self,
99 function_name: str,
100 user_id: str,
101 prompt: str = "",
102 variables: Optional[dict[str, Any]] = None,
103 include: Optional[IncludeOptions] = None,
104 ) -> Function:
105 """
106 Gets a function by name.
108 Args:
109 function_name: The name of the function to get.
110 user_id: The ID of the user.
111 prompt: The prompt to use for predicting the function's arguments.
112 Omit if you don't want to predict the arguments.
113 variables: The variables to use for auto-mapping the arguments.
114 Omit if you don't want to auto-map the arguments.
115 include: The fields to include with the function.
117 Returns:
118 The function.
119 """
120 query_string = self._get_query_string(include, None, None, None)
122 body: dict[str, Any] = {}
123 if prompt:
124 body["prompt"] = prompt
126 if variables:
127 body["_variables"] = variables
129 response = await self.http_client.post(
130 f"{self.name}/{function_name}/get/{query_string}",
131 headers=self._get_signed_request_headers(user_id),
132 json=body,
133 )
134 data = self._get_response_data_or_raise(response)
136 function = Function(**data, _resource=self)
137 return function
139 async def call(
140 self,
141 function_name: str,
142 arguments: dict[str, Any],
143 user_id: str,
144 variables: Optional[dict[str, Any]] = None,
145 connected_account_id: Optional[int] = None,
146 ) -> FunctionCallOutput:
147 """
148 Calls a function with the given arguments and variables.
150 Args:
151 function_name: The name of the function to call.
152 arguments: Values for the function's parameters.
153 user_id: The user ID of the user on whose behalf the function will be called.
154 variables: The variables to pass to the function, if any.
156 Returns:
157 The function's output.
158 """
160 if "cursor" in arguments:
161 # LangChain doesn't support aliases in the arguments schema, so we
162 # handle the cursor parameter.
163 # TODO: Remove this once LangChain supports aliases in the arguments schema.
164 arguments["_cursor"] = arguments["cursor"]
166 url = f"{self.name}/{function_name}/call/"
167 if connected_account_id:
168 url += f"?connected_account_id={connected_account_id}"
170 response = await self.http_client.post(
171 url,
172 headers=self._get_signed_request_headers(user_id),
173 json={**arguments, "_variables": variables},
174 )
175 if response.status_code == 400:
176 self._raise_function_call_exception(response)
178 data = self._get_response_data_or_raise(response)
180 if "_cursor" in data:
181 return PaginatedFunctionCallOutput(**data)
183 return FunctionCallOutput(**data)
185 def call_sync(
186 self,
187 function_name: str,
188 arguments: dict[str, Any],
189 user_id: str,
190 variables: Optional[dict[str, Any]] = None,
191 ) -> FunctionCallOutput:
192 """
193 Calls a function synchronously with the given arguments and variables.
195 Args:
196 function_name: The name of the function to call.
197 arguments: Values for the function's parameters.
198 user_id: The user ID of the user on whose behalf the function will be called.
199 variables: The variables to pass to the function, if any.
201 Returns:
202 The function's output.
203 """
205 if "cursor" in arguments:
206 # LangChain doesn't support aliases in the arguments schema, so we
207 # handle the cursor parameter.
208 # TODO: Remove this once LangChain supports aliases in the arguments schema.
209 arguments["_cursor"] = arguments["cursor"]
211 response = httpx.post(
212 f"{self.http_client.base_url}/{self.name}/{function_name}/call/",
213 headers=self._get_signed_request_headers(user_id),
214 json={**arguments, "_variables": variables},
215 )
216 if response.status_code == 400:
217 self._raise_function_call_exception(response)
219 data = self._get_response_data_or_raise(response)
221 if "_cursor" in data:
222 return PaginatedFunctionCallOutput(**data)
224 return FunctionCallOutput(**data)
226 def _raise_function_call_exception(self, response: Any):
227 data = response.json()
228 error_details = data.get("error_details")
229 error_message = "Failed to call the function."
230 errors = []
232 if isinstance(error_details, dict):
233 errors = [f"{key}: {value}" for key, value in error_details.items()]
235 elif isinstance(error_details, list):
236 errors = error_details
238 error_message += f"\nDetails: {', '.join(errors)}"
240 raise FunctionCallError(error_message, errors=errors)
242 @staticmethod
243 def _get_query_string(
244 include: Optional[IncludeOptions],
245 app: str | None,
246 connected_only: bool | None,
247 type: FunctionType | None,
248 predict_arguments: bool = False,
249 ) -> str:
250 query_params: dict[str, Any] = {}
251 if include:
252 query_params["include"] = ",".join(include)
253 if app:
254 query_params["app"] = app
255 if connected_only:
256 query_params["connected_only"] = connected_only
257 if type:
258 query_params["type"] = type
260 if predict_arguments:
261 query_params["predict_arguments"] = "true" if predict_arguments else "false"
263 query_string = urlencode(query_params, doseq=False)
264 if query_string:
265 query_string = f"?{query_string}"
267 return query_string