Repository navigation
Expand file tree
/
Copy pathhttp_plugin.py
More file actions
145 lines (112 loc) · 4.98 KB
/
Copy pathhttp_plugin.py
File metadata and controls
145 lines (112 loc) · 4.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
# Copyright (c) Microsoft. All rights reserved.
import json
from typing import Annotated, Any
from urllib.parse import urlparse
import aiohttp
from semantic_kernel.exceptions import FunctionExecutionException
from semantic_kernel.functions.kernel_function_decorator import kernel_function
from semantic_kernel.kernel_pydantic import KernelBaseModel
class HttpPlugin(KernelBaseModel):
"""A plugin that provides HTTP functionality.
Usage:
kernel.add_plugin(HttpPlugin(), "http")
# With allowed domains for security:
kernel.add_plugin(HttpPlugin(allowed_domains=["example.com", "api.example.com"]), "http")
Examples:
{{http.getAsync $url}}
{{http.postAsync $url}}
{{http.putAsync $url}}
{{http.deleteAsync $url}}
"""
allowed_domains: set[str] | None = None
"""List of allowed domains to send requests to. If None, all domains are allowed."""
def _is_uri_allowed(self, url: str) -> bool:
"""Check if the URL's host is in the allowed domains list.
Args:
url: The URL to check.
Returns:
True if the URL is allowed, False otherwise.
"""
if self.allowed_domains is None:
return True
parsed = urlparse(url)
host = parsed.hostname
if host is None:
return False
# Case-insensitive comparison
return host.lower() in {domain.lower() for domain in self.allowed_domains}
def _validate_url(self, url: str) -> None:
"""Validate the URL, checking if it's not empty and is in the allowed domains.
Args:
url: The URL to validate.
Raises:
FunctionExecutionException: If the URL is empty or not in the allowed domains.
"""
if not url:
raise FunctionExecutionException("url cannot be `None` or empty")
if not self._is_uri_allowed(url):
raise FunctionExecutionException("Sending requests to the provided location is not allowed.")
@kernel_function(description="Makes a GET request to a url", name="getAsync")
async def get(self, url: Annotated[str, "The URL to send the request to."]) -> str:
"""Sends an HTTP GET request to the specified URI and returns the response body as a string.
Args:
url: The URL to send the request to.
Returns:
The response body as a string.
"""
self._validate_url(url)
async with aiohttp.ClientSession() as session, session.get(url, raise_for_status=True) as response:
return await response.text()
@kernel_function(description="Makes a POST request to a uri", name="postAsync")
async def post(
self,
url: Annotated[str, "The URI to send the request to."],
body: Annotated[dict[str, Any] | None, "The body of the request"] = None,
) -> str:
"""Sends an HTTP POST request to the specified URI and returns the response body as a string.
Args:
url: The URI to send the request to.
body: Contains the body of the request
returns:
The response body as a string.
"""
self._validate_url(url)
headers = {"Content-Type": "application/json"}
data = json.dumps(body) if body is not None else None
async with (
aiohttp.ClientSession() as session,
session.post(url, headers=headers, data=data, raise_for_status=True) as response,
):
return await response.text()
@kernel_function(description="Makes a PUT request to a uri", name="putAsync")
async def put(
self,
url: Annotated[str, "The URI to send the request to."],
body: Annotated[dict[str, Any] | None, "The body of the request"] = None,
) -> str:
"""Sends an HTTP PUT request to the specified URI and returns the response body as a string.
Args:
url: The URI to send the request to.
body: Contains the body of the request
Returns:
The response body as a string.
"""
self._validate_url(url)
headers = {"Content-Type": "application/json"}
data = json.dumps(body) if body is not None else None
async with (
aiohttp.ClientSession() as session,
session.put(url, headers=headers, data=data, raise_for_status=True) as response,
):
return await response.text()
@kernel_function(description="Makes a DELETE request to a uri", name="deleteAsync")
async def delete(self, url: Annotated[str, "The URI to send the request to."]) -> str:
"""Sends an HTTP DELETE request to the specified URI and returns the response body as a string.
Args:
url: The URI to send the request to.
Returns:
The response body as a string.
"""
self._validate_url(url)
async with aiohttp.ClientSession() as session, session.delete(url, raise_for_status=True) as response:
return await response.text()