Spaces:
Sleeping
Sleeping
File size: 2,766 Bytes
01d5a5d |
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 |
from typing import Any
from mcp.server.fastmcp import FastMCP
import http.client
import json
import requests
mindverse = FastMCP("mindverse_public")
url = "app.secondme.io"
messages =[]
@mindverse.tool()
async def get_response(query:str, instance_id:str) -> str | None:
"""
Received a response based on public secondme model.
Args:
query (str): Questions raised by users regarding the secondme model.
instance_id (str): ID used to identify the secondme model, or url used to identify the secondme model.
"""
id = instance_id.split('/')[-1]
path = f"/api/chat/{id}"
headers = {"Content-Type": "application/json"}
messages.append({"role": "user", "content": query})
data = {
"messages": messages,
"metadata": {
"enable_l0_retrieval": False,
"role_id": "default_role"
},
"temperature": 0.7,
"max_tokens": 2000,
"stream": True
}
conn = http.client.HTTPSConnection(url)
# Send the POST request
conn.request("POST", path, body=json.dumps(data), headers=headers)
# Get the response
response = conn.getresponse()
full_content = ""
for line in response:
if line:
decoded_line = line.decode('utf-8').strip()
if decoded_line == 'data: [DONE]':
break
if decoded_line.startswith('data: '):
try:
json_str = decoded_line[6:]
chunk = json.loads(json_str)
content = chunk['choices'][0]['delta'].get('content', '')
if content:
full_content += content
except json.JSONDecodeError:
pass
conn.close()
if full_content:
messages.append({"role": "system", "content": full_content})
return full_content
else:
return None
@mindverse.tool()
async def get_online_instances():
"""
Check which secondme models are available for chatting online.
"""
url = "https://app.secondme.io/api/upload/list?page_size=100"
response = requests.get(url)
if response.status_code == 200:
data = response.json()
items = data.get("data", {}).get("items", [])
online_items = [
{
"upload_name": item["upload_name"],
"instance_id": item["instance_id"],
"description": item["description"]
}
for item in items if item.get("status") == "online"
]
return json.dumps(online_items, ensure_ascii=False, indent=2)
else:
raise Exception(f"Request failed with status code: {response.status_code}")
if __name__ == "__main__":
mindverse.run(transport='stdio')
|