-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
178 lines (145 loc) · 6.8 KB
/
main.py
File metadata and controls
178 lines (145 loc) · 6.8 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any, Optional
import httpx
import json
from mcp.server.fastmcp import FastMCP
from os import getenv
from dotenv import load_dotenv
async def get_game_details(
game_name: str,
fields: Optional[list[str]] = None,
include: Optional[list[str]] = None
) -> dict:
"""
Search through games on TheGamesDB by name. Returns: JSON array. Argument: name (required).
"""
"""
Makes an asynchronous GET request to TheGamesDB API to retrieve game details by name.
Args:
game_name (str): The name of the game to search for.
fields (list[str], optional): A list of additional fields to include in the response.
Defaults to ['players', 'publishers', 'genres', 'overview',
'last_updated', 'rating', 'platform', 'coop',
'youtube', 'os', 'processor', 'ram', 'hdd',
'video', 'sound', 'alternates'].
include (list[str], optional): A list of related data to include (e.g., 'boxart', 'platform').
Defaults to ['boxart', 'platform'].
Returns:
dict: A dictionary containing the JSON response from the API, or an error message.
"""
base_url = "https://api.thegamesdb.net/v1.1/Games/ByGameName"
# Set default values for fields and include if not provided
if fields is None:
fields = [
'players', 'publishers', 'genres', 'overview', 'last_updated', 'rating',
'platform', 'coop', 'youtube', 'os', 'processor', 'ram', 'hdd',
'video', 'sound', 'alternates'
]
if include is None:
include = ['boxart', 'platform']
# Construct query parameters
params = {
"apikey": THEGAMESDB_API_KEY, # Use the global API key
"name": game_name,
"fields": ",".join(fields), # Join list elements with a comma
"include": ",".join(include) # Join list elements with a comma
}
try:
# Use an asynchronous httpx client
async with httpx.AsyncClient() as client:
print(f"Making request to: {base_url} with params: {params}")
response = await client.get(base_url, params=params, follow_redirects=True)
# Raise an exception for bad status codes (4xx or 5xx)
response.raise_for_status()
# Parse the JSON response
data = response.json()
return data
except httpx.RequestError as e:
print(f"An error occurred while requesting {e.request.url!r}: {e}")
return {"error": f"Network or request error: {e}"}
except httpx.HTTPStatusError as e:
print(f"Error response {e.response.status_code} while requesting {e.request.url!r}: {e}")
return {"error": f"API returned an error: Status {e.response.status_code}, Response: {e.response.text}"}
except json.JSONDecodeError as e:
print(f"Failed to decode JSON response: {e}")
return {"error": f"Invalid JSON response from API: {e}"}
except Exception as e:
print(f"An unexpected error occurred: {e}")
return {"error": f"An unexpected error occurred: {e}"}
async def get_platform_details() -> dict:
"""
Makes an asynchronous GET request to TheGamesDB API to retrieve platform details.
Args:
fields (list[str], optional): A list of additional fields to include in the response.
Defaults to ['icon', 'console', 'controller', 'developer',
'manufacturer', 'media', 'cpu', 'memory',
'graphics', 'sound', 'maxcontrollers',
'display', 'overview', 'youtube'].
Returns:
dict: A dictionary containing the JSON response from the API, or an error message.
"""
base_url = "https://api.thegamesdb.net/v1/Platforms"
# Set default values for fields if not provided
fields = [
'icon', 'console', 'controller', 'developer', 'manufacturer', 'media',
'cpu', 'memory', 'graphics', 'sound', 'maxcontrollers', 'display',
'overview', 'youtube'
]
# Construct query parameters
params = {
"apikey": THEGAMESDB_API_KEY, # Use the global API key loaded from .env
"fields": ",".join(fields), # Join list elements with a comma
}
try:
# Use an asynchronous httpx client
async with httpx.AsyncClient() as client:
print(f"Making request to: {base_url} with params: {params}")
response = await client.get(base_url, params=params, follow_redirects=True)
# Raise an exception for bad status codes (4xx or 5xx)
response.raise_for_status()
# Parse the JSON response
data = response.json()
return data
except httpx.RequestError as e:
print(f"An error occurred while requesting {e.request.url!r}: {e}")
return {"error": f"Network or request error: {e}"}
except httpx.HTTPStatusError as e:
print(f"Error response {e.response.status_code} while requesting {e.request.url!r}: {e}")
return {"error": f"API returned an error: Status {e.response.status_code}, Response: {e.response.text}"}
except json.JSONDecodeError as e:
print(f"Failed to decode JSON response: {e}")
return {"error": f"Invalid JSON response from API: {e}"}
except Exception as e:
print(f"An unexpected error occurred: {e}")
return {"error": f"An unexpected error occurred: {e}"}
# Initialize FastMCP server
mcp = FastMCP("testmonkey")
@mcp.tool()
async def TGDB_search_game(game_name: str) -> str:
a = await get_game_details(game_name)
return json.dumps(a)
@mcp.tool()
async def TGDB_platforms_details() -> str:
"""
Returns the Platforms (C64, Amiga etc) recorded on TheGamesDB. Returns: JSON data.
"""
a = await get_platform_details()
return json.dumps(a)
# Run the server
if __name__ == "__main__":
# Load environment variables from a .env file
# Ensure you have 'python-dotenv' installed (pip install python-dotenv)
# And create a .env file in the same directory as your script with:
# THEGAMESDB_API_KEY="YOUR_ACTUAL_API_KEY_HERE"
load_dotenv()
# Define the API key globally, loaded from environment variables.
# This is a more secure way to manage API keys than hardcoding them.
THEGAMESDB_API_KEY = getenv("THEGAMESDB_API_KEY")
# Ensure API key is available before making the request
if not THEGAMESDB_API_KEY:
print ({"error": "API Key is missing. Cannot proceed with the request."})
exit()
# Initialize and run the server
mcp.run(transport='stdio')