Add custom tools (#2292)
Co-authored-by: luowei <glpat-EjySCyNjWiLqAED-YmwM> Co-authored-by: crazywoola <427733928@qq.com> Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com>pull/2293/head
parent
bf3ee660e0
commit
d2797abdb4
@ -0,0 +1,61 @@
|
|||||||
|
import json
|
||||||
|
import requests
|
||||||
|
from datetime import datetime
|
||||||
|
from urllib.parse import quote
|
||||||
|
from core.tools.tool.builtin_tool import BuiltinTool
|
||||||
|
from core.tools.entities.tool_entities import ToolInvokeMessage
|
||||||
|
|
||||||
|
from typing import Any, Dict, List, Union
|
||||||
|
|
||||||
|
|
||||||
|
class GihubRepositoriesTool(BuiltinTool):
|
||||||
|
def _invoke(self, user_id: str, tool_paramters: Dict[str, Any]) -> Union[ToolInvokeMessage, List[ToolInvokeMessage]]:
|
||||||
|
"""
|
||||||
|
invoke tools
|
||||||
|
"""
|
||||||
|
top_n = tool_paramters.get('top_n', 5)
|
||||||
|
query = tool_paramters.get('query', '')
|
||||||
|
if not query:
|
||||||
|
return self.create_text_message('Please input symbol')
|
||||||
|
|
||||||
|
if 'access_tokens' not in self.runtime.credentials or not self.runtime.credentials.get('access_tokens'):
|
||||||
|
return self.create_text_message("Github API Access Tokens is required.")
|
||||||
|
if 'api_version' not in self.runtime.credentials or not self.runtime.credentials.get('api_version'):
|
||||||
|
api_version = '2022-11-28'
|
||||||
|
else:
|
||||||
|
api_version = self.runtime.credentials.get('api_version')
|
||||||
|
|
||||||
|
try:
|
||||||
|
headers = {
|
||||||
|
"Content-Type": "application/vnd.github+json",
|
||||||
|
"Authorization": f"Bearer {self.runtime.credentials.get('access_tokens')}",
|
||||||
|
"X-GitHub-Api-Version": api_version
|
||||||
|
}
|
||||||
|
s = requests.session()
|
||||||
|
api_domain = 'https://api.github.com'
|
||||||
|
response = s.request(method='GET', headers=headers,
|
||||||
|
url=f"{api_domain}/search/repositories?"
|
||||||
|
f"q={quote(query)}&sort=stars&per_page={top_n}&order=desc")
|
||||||
|
response_data = response.json()
|
||||||
|
if response.status_code == 200 and isinstance(response_data.get('items'), list):
|
||||||
|
contents = list()
|
||||||
|
if len(response_data.get('items')) > 0:
|
||||||
|
for item in response_data.get('items'):
|
||||||
|
content = dict()
|
||||||
|
updated_at_object = datetime.strptime(item['updated_at'], "%Y-%m-%dT%H:%M:%SZ")
|
||||||
|
content['owner'] = item['owner']['login']
|
||||||
|
content['name'] = item['name']
|
||||||
|
content['description'] = item['description'][:100] + '...' if len(item['description']) > 100 else item['description']
|
||||||
|
content['url'] = item['html_url']
|
||||||
|
content['star'] = item['watchers']
|
||||||
|
content['forks'] = item['forks']
|
||||||
|
content['updated'] = updated_at_object.strftime("%Y-%m-%d")
|
||||||
|
contents.append(content)
|
||||||
|
s.close()
|
||||||
|
return self.create_text_message(self.summary(user_id=user_id, content=json.dumps(contents, ensure_ascii=False)))
|
||||||
|
else:
|
||||||
|
return self.create_text_message(f'No items related to {query} were found.')
|
||||||
|
else:
|
||||||
|
return self.create_text_message((response.json()).get('message'))
|
||||||
|
except Exception as e:
|
||||||
|
return self.create_text_message("Github API Key and Api Version is invalid. {}".format(e))
|
||||||
Loading…
Reference in New Issue