samsonleegh commited on
Commit
411df6f
·
verified ·
1 Parent(s): 240e2c3

Upload 2 files

Browse files
Files changed (2) hide show
  1. cryptopanic_news.py +44 -0
  2. cryptopanic_news_server.py +47 -0
cryptopanic_news.py ADDED
@@ -0,0 +1,44 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import requests
2
+
3
+ def get_posts(
4
+ auth_token: str,
5
+ kind: str = None, # e.g. "news", "blogs", "videos"
6
+ currencies: list = None, # e.g. ["BTC", "ETH"]
7
+ regions: list = None, # e.g. ["en"] or country codes
8
+ public: bool = True, # whether to include public posts
9
+ filter_duplicates: bool = True,
10
+ page: int = 1,
11
+ filter: str = None # a general filter string
12
+ ) -> dict:
13
+ """
14
+ Fetch posts from CryptoPanic.
15
+ """
16
+ API_BASE = "https://cryptopanic.com/api/developer/v2"
17
+ #url = f"{API_BASE}/posts/"
18
+ url = "https://cryptopanic.com/api/developer/v2/posts/"
19
+ headers = {
20
+ "Accept": "application/json"
21
+ }
22
+ params = {
23
+ "auth_token": auth_token,
24
+ "public": "true" if public else "false",
25
+ "page": page,
26
+ }
27
+ if kind:
28
+ params["kind"] = kind
29
+ if currencies:
30
+ print(currencies)
31
+ # CryptoPanic expects multiple items comma-separated or repeated parameters
32
+ # Checking docs: uses comma-separated list.
33
+ params["currencies"] = ",".join(currencies)
34
+ if regions:
35
+ params["regions"] = ",".join(regions)
36
+ if filter_duplicates:
37
+ params["filter_duplicates"] = "true"
38
+ if filter:
39
+ params["filter"] = filter
40
+
41
+ response = requests.get(url, params=params, headers=headers)
42
+ response.raise_for_status()
43
+ print(url)
44
+ return response.json()
cryptopanic_news_server.py ADDED
@@ -0,0 +1,47 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from mcp.server.fastmcp import FastMCP
2
+ from cryptopanic_news import get_posts
3
+ import os
4
+ from dotenv import load_dotenv
5
+ load_dotenv(override=True)
6
+
7
+ mcp = FastMCP("cryptopanic_news_server")
8
+
9
+ @mcp.tool()
10
+ async def get_crypto_news(currencies_ls: list | str | None = None) -> str:
11
+ """Fetch latest cryptocurrency news from CryptoPanic.
12
+
13
+ Args:
14
+ currencies_ls: The list of cryptocurrency symbols to filter news by. Accepts a single string (e.g. "BTC") or a list of symbols (e.g. ["HYPE", "ETC", "BTC", "XRP"]).
15
+
16
+ If set to None, retrieves the general CryptoPanic news feed
17
+ without filtering by any specific currency.
18
+ """
19
+ token = os.getenv("CRYPTOPANIC_API_KEY")
20
+ if not token:
21
+ return "Error: CRYPTOPANIC_API_KEY not set."
22
+
23
+ # Allow a single string or a list
24
+ if isinstance(currencies_ls, str):
25
+ currencies_ls = [currencies_ls]
26
+
27
+ data = get_posts(
28
+ auth_token=token,
29
+ kind="news",
30
+ currencies=currencies_ls,
31
+ page=1
32
+ )
33
+
34
+ if not isinstance(data, dict):
35
+ return f"Error: Unexpected API response: {data!r}"
36
+
37
+ results = []
38
+ for post in data.get("results", []):
39
+ title = post.get("title") or "No title"
40
+ desc = post.get("description") or ""
41
+ pub = post.get("published_at") or "Unknown time"
42
+ results.append(f"title: {title}\n{desc[:200]}\npublished at: {pub}\n")
43
+
44
+ return "\n".join(results) if results else "No recent crypto news found."
45
+
46
+ if __name__ == "__main__":
47
+ mcp.run(transport='stdio')