|
| 1 | +# TODO: Remove when Python 3.9 support is dropped |
| 2 | +from __future__ import annotations |
| 3 | + |
1 | 4 | import asyncio |
2 | 5 | import base64 |
3 | 6 | import json |
4 | 7 | from collections.abc import Sequence |
5 | 8 | from enum import Enum |
6 | 9 | from functools import partial |
7 | | -from typing import Annotated, Any, TypeAlias, cast |
| 10 | +from typing import Annotated, Any, cast |
| 11 | +from urllib.parse import quote |
8 | 12 |
|
9 | 13 | import requests |
10 | 14 | from langchain_core.tools import BaseTool |
11 | 15 | from pydantic import BaseModel, BeforeValidator, Field, PrivateAttr |
12 | 16 | from requests.exceptions import RequestException |
13 | 17 |
|
| 18 | +# TODO: Remove when Python 3.9 support is dropped |
| 19 | +from typing_extensions import TypeAlias |
| 20 | + |
14 | 21 | # Type aliases for common types |
15 | 22 | JsonDict: TypeAlias = dict[str, Any] |
16 | 23 | Headers: TypeAlias = dict[str, str] |
@@ -140,21 +147,24 @@ def _prepare_request_params(self, kwargs: JsonDict) -> tuple[str, JsonDict, Json |
140 | 147 | for key, value in kwargs.items(): |
141 | 148 | param_location = self._execute_config.parameter_locations.get(key) |
142 | 149 |
|
143 | | - match param_location: |
144 | | - case ParameterLocation.PATH: |
145 | | - url = url.replace(f"{{{key}}}", str(value)) |
146 | | - case ParameterLocation.QUERY: |
| 150 | + if param_location == ParameterLocation.PATH: |
| 151 | + # Safely encode path parameters to prevent SSRF attacks |
| 152 | + encoded_value = quote(str(value), safe="") |
| 153 | + url = url.replace(f"{{{key}}}", encoded_value) |
| 154 | + elif param_location == ParameterLocation.QUERY: |
| 155 | + query_params[key] = value |
| 156 | + elif param_location in (ParameterLocation.BODY, ParameterLocation.FILE): |
| 157 | + body_params[key] = value |
| 158 | + else: |
| 159 | + # Default behavior |
| 160 | + if f"{{{key}}}" in url: |
| 161 | + # Safely encode path parameters to prevent SSRF attacks |
| 162 | + encoded_value = quote(str(value), safe="") |
| 163 | + url = url.replace(f"{{{key}}}", encoded_value) |
| 164 | + elif self._execute_config.method in {"GET", "DELETE"}: |
147 | 165 | query_params[key] = value |
148 | | - case ParameterLocation.BODY | ParameterLocation.FILE: |
| 166 | + else: |
149 | 167 | body_params[key] = value |
150 | | - case _: |
151 | | - # Default behavior |
152 | | - if f"{{{key}}}" in url: |
153 | | - url = url.replace(f"{{{key}}}", str(value)) |
154 | | - elif self._execute_config.method in {"GET", "DELETE"}: |
155 | | - query_params[key] = value |
156 | | - else: |
157 | | - body_params[key] = value |
158 | 168 |
|
159 | 169 | return url, body_params, query_params |
160 | 170 |
|
@@ -355,13 +365,12 @@ def to_langchain(self) -> BaseTool: |
355 | 365 | python_type: type = str # Default to str |
356 | 366 | if isinstance(details, dict): |
357 | 367 | type_str = details.get("type", "string") |
358 | | - match type_str: |
359 | | - case "number": |
360 | | - python_type = float |
361 | | - case "integer": |
362 | | - python_type = int |
363 | | - case "boolean": |
364 | | - python_type = bool |
| 368 | + if type_str == "number": |
| 369 | + python_type = float |
| 370 | + elif type_str == "integer": |
| 371 | + python_type = int |
| 372 | + elif type_str == "boolean": |
| 373 | + python_type = bool |
365 | 374 |
|
366 | 375 | field = Field(description=details.get("description", "")) |
367 | 376 | else: |
@@ -480,7 +489,7 @@ def to_langchain(self) -> Sequence[BaseTool]: |
480 | 489 | """ |
481 | 490 | return [tool.to_langchain() for tool in self.tools] |
482 | 491 |
|
483 | | - def meta_tools(self) -> "Tools": |
| 492 | + def meta_tools(self) -> Tools: |
484 | 493 | """Return meta tools for tool discovery and execution |
485 | 494 |
|
486 | 495 | Meta tools enable dynamic tool discovery and execution based on natural language queries. |
|
0 commit comments