EveryInfra API 및 MCP 문서

import os

import requests

EVERYINFRA_API_KEY = os.environ["EVERYINFRA_API_KEY"]
IPFOXY_PROXY = os.environ["IPFOXY_PROXY"]  # http://username:password@host:port
PROXIES = {"http": IPFOXY_PROXY, "https": IPFOXY_PROXY}

# 1. 직접 보내는 요청은 IPFoxy 출구 IP를 거쳐 나갑니다.
exit_ip = requests.get("https://ipinfo.io/json", proxies=PROXIES, timeout=30).json()
print("Exit IP country:", exit_ip.get("country"))

# 2. EveryInfra 호출은 직접 전송되며 프록시가 필요하지 않습니다.
direct = requests.Session()
direct.trust_env = False  # HTTP(S)_PROXY를 무시하므로 이 호출은 프록시 트래픽을 쓰지 않습니다
resp = direct.post(
    "https://api.everyinfra.com/api/v1/social",
    headers={"Authorization": f"Bearer {EVERYINFRA_API_KEY}"},
    json={
        "platform": "amazon",
        "action": "profile",
        "params": {"url": "https://www.amazon.com/dp/B0CYJDVT31"},
    },
    timeout=120,
)
resp.raise_for_status()
body = resp.json()
item = body["results"]
print(item["title"][:60], item["currency"], item["price"], item["rating"])
print("Charged USD:", body["billing"]["charged_usd"])

import os
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlsplit

import requests

EXA_API_KEY = os.environ["EXA_API_KEY"]
EVERYINFRA_API_KEY = os.environ["EVERYINFRA_API_KEY"]
QUESTION = "note-taking apps similar to Obsidian"

# 1. Exa: 키워드가 아닌 의미로 기업 찾기
companies = requests.post(
    "https://api.exa.ai/search",
    headers={"x-api-key": EXA_API_KEY},
    json={"query": QUESTION, "category": "company", "type": "auto", "numResults": 10},
    timeout=60,
).json()["results"]


# 2. EveryInfra: 기업별 G2 평점과 리뷰 수
def g2_rating(company):
    name = company["title"].split(" - ")[0].split(" | ")[0].strip()
    resp = requests.post(
        "https://api.everyinfra.com/api/v1/social",
        headers={"Authorization": f"Bearer {EVERYINFRA_API_KEY}"},
        json={"platform": "g2", "action": "search", "params": {"keyword": name}},
        timeout=180,
    )
    resp.raise_for_status()
    body = resp.json()
    match = next((p for p in body["results"] if p["product_name"].lower() == name.lower()), None)
    return name, match, body["billing"]["charged_usd"]


with ThreadPoolExecutor(max_workers=4) as pool:
    rows = list(pool.map(g2_rating, companies))

spent = 0.0
for company, (name, g2, charged) in zip(companies, rows):
    spent += charged
    domain = urlsplit(company["url"]).hostname.removeprefix("www.")
    if g2 and g2["review_count"]:
        print(f"{name:12} {domain:16} G2 {g2['rating']} / 5 · {g2['review_count']} reviews")
    else:
        print(f"{name:12} {domain:16} no G2 reviews yet")
print("EveryInfra charged (USD):", round(spent, 6))

import os

import requests

EVERYINFRA_API_KEY = os.environ["EVERYINFRA_API_KEY"]
MONID_API_KEY = os.environ["MONID_API_KEY"]
MONID = "https://api.monid.ai/v1"
MONID_HEADERS = {"Authorization": f"Bearer {MONID_API_KEY}"}
MAX_PRICE_USD = 0.01  # do not run a Monid tool that costs more than this per call

# 1. The daily job: EveryInfra xiaohongshu/search, one fixed request shape
resp = requests.post(
    "https://api.everyinfra.com/api/v1/social",
    headers={"Authorization": f"Bearer {EVERYINFRA_API_KEY}"},
    json={"platform": "xiaohongshu", "action": "search",
          "params": {"keyword": "matcha latte", "sort": "most_liked"}},
    timeout=120,
)
resp.raise_for_status()
body = resp.json()
for note in body["results"][:3]:
    print(note["like_count"], note["collect_count"], note["posted_at"][:10], note["url"])
print("EveryInfra:", len(body["results"]), "notes, charged (USD):", body["billing"]["charged_usd"])

# 2. A one-off lookup that is not a daily platform: discover, inspect, run on Monid
found = requests.post(f"{MONID}/discover", headers=MONID_HEADERS, timeout=60,
                      json={"query": "company logo image URL from a domain", "limit": 3}).json()["results"]
# discover adds ranking and price fields to each result; inspect and run take only the fields that name the tool
SHOWN_ONLY = ("providerName", "description", "price", "score", "scoreBreakdown", "tags", "metrics", "hints")
tool = {key: value for key, value in found[0].items() if key not in SHOWN_ONLY}
detail = requests.post(f"{MONID}/inspect", headers=MONID_HEADERS, timeout=60, json=tool).json()
price = detail["price"]["amount"]["value"]
print("Monid tool:", detail["endpoint"], "price (USD):", price)
if price <= MAX_PRICE_USD:
    run = requests.post(f"{MONID}/run", headers=MONID_HEADERS, timeout=120,
                        json={**tool, "input": {"pathParams": {"company": "starbucks.com"}}}).json()
    print("Monid run:", run.get("runId"), run.get("status"), run.get("output"))