Build a Python Client
Create a small Python client for Y2 JSON reads, writes, errors, and pagination
This guide builds a dependency-free Python client around Y2's public HTTP contract. Use it directly for small services and jobs, or use the same behaviors as acceptance criteria for a generated client.
Keep credentials out of notebooks and source files
Load Y2_API_KEY from the environment or a secret manager. Notebook output, exception dumps,
and committed .env files can leak credentials as easily as application logs.
1. Configure the environment
The examples use Python 3.11 or newer and only the standard library:
export Y2_API_KEY="y2_..."
python --versionNew keys require a Pro or Elite workspace and the scopes needed by the operations you call.
2. Create the transport
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from typing import Any, Mapping, TypeVar, cast
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
T = TypeVar("T")
API_ORIGIN = "https://api.y2.dev"
API_V1 = f"{API_ORIGIN}/api/v1"
class Y2ApiError(Exception):
def __init__(
self,
status: int,
problem: dict[str, Any] | None,
retry_after: str | None,
request_id: str | None,
) -> None:
detail = problem.get("detail") if problem else None
super().__init__(detail or f"Y2 request failed with HTTP {status}")
self.status = status
self.problem = problem
self.retry_after = retry_after
self.request_id = request_id
@dataclass(frozen=True)
class Y2Client:
api_key: str
timeout_seconds: float = 30.0
@classmethod
def from_env(cls) -> Y2Client:
api_key = os.environ.get("Y2_API_KEY")
if not api_key:
raise RuntimeError("Y2_API_KEY is required")
return cls(api_key=api_key)
def request_json(
self,
method: str,
path_or_url: str,
*,
query: Mapping[str, str | int | float | bool | None] | None = None,
body: Mapping[str, Any] | None = None,
headers: Mapping[str, str] | None = None,
) -> T:
if path_or_url.startswith("https://"):
url = path_or_url
elif path_or_url.startswith("/api/"):
url = f"{API_ORIGIN}{path_or_url}"
else:
url = f"{API_V1}{path_or_url}"
params = {
key: str(value).lower() if isinstance(value, bool) else value
for key, value in (query or {}).items()
if value is not None
}
if params:
separator = "&" if "?" in url else "?"
url = f"{url}{separator}{urlencode(params)}"
request_headers = {
"Authorization": f"Bearer {self.api_key}",
"Accept": "application/json",
**(headers or {}),
}
data = None
if body is not None:
request_headers.setdefault("Content-Type", "application/json")
data = json.dumps(body, separators=(",", ":")).encode("utf-8")
request = Request(url, data=data, headers=request_headers, method=method)
try:
with urlopen(request, timeout=self.timeout_seconds) as response:
raw = response.read()
if response.status == 204 or not raw:
return cast(T, None)
return cast(T, json.loads(raw))
except HTTPError as error:
raw = error.read()
problem = None
if raw:
try:
decoded = json.loads(raw)
if isinstance(decoded, dict):
problem = decoded
except json.JSONDecodeError:
pass
raise Y2ApiError(
error.code,
problem,
error.headers.get("Retry-After"),
error.headers.get("X-Request-Id"),
) from errorThis helper handles JSON and 204 No Content. Create separate response readers for Markdown, plain
text, NDJSON, GeoJSON, audio redirects, and SSE instead of forcing every media type through
json.loads.
3. Read current News items
from typing import Any, TypedDict
from y2_client import Y2Client
class NewsPage(TypedDict):
data: list[dict[str, Any]]
meta: dict[str, Any]
links: dict[str, str | None]
client = Y2Client.from_env()
page: NewsPage = client.request_json(
"GET",
"/news",
query={"topics": "markets,macro", "limit": 10},
)
for item in page["data"]:
print(item["title"], item["url"])Use GET /news/feeds to discover the current 40-topic catalog. Current items use title, summary,
and a nested sentiment.label/sentiment.value object—not older top-level signal and sentiment
scalar fields.
4. Create a profile safely
from typing import Any
from y2_client import Y2Client
client = Y2Client.from_env()
created: dict[str, Any] = client.request_json(
"POST",
"/profiles",
headers={"Idempotency-Key": "supplier-risk-2026-07-21"},
body={
"name": "Critical Supplier Risk",
"topic": "Monitor disruption affecting strategic semiconductor suppliers.",
"frequency": "daily",
"scheduleTimeOfDay": "08:00",
},
)
profile_id = created["data"]["id"]
print(profile_id) # prf_...Creation returns the profile in data. It also creates an active subscription, but retrieve that
separate sub_... resource from GET /profiles when you need to change delivery.
Use PATCH /profiles/{profileId} for partial changes. PUT replaces the complete mutable state,
and omitted optional values reset or clear. Retain returned ETag headers if your workflow uses
If-Match; a JSON-only wrapper that discards headers should be extended for that workflow.
5. Follow report pagination
from collections.abc import Iterator
from typing import Any
from y2_client import Y2Client
def reports_for_profile(
client: Y2Client,
profile_id: str,
) -> Iterator[dict[str, Any]]:
next_url: str | None = "/reports"
first = True
while next_url:
page: dict[str, Any] = client.request_json(
"GET",
next_url,
query={"profileId": profile_id, "limit": 20} if first else None,
)
yield from page["data"]
next_url = page["links"]["next"]
first = False
client = Y2Client.from_env()
for report in reports_for_profile(client, "prf_0123456789abcdef01234567"):
print(report["id"], report["summary"])Treat links.next and its cursor as opaque. Do not decode, edit, or combine a continuation link
with different filters.
6. Run independent reads concurrently
The standard-library transport is synchronous. For a small job, run independent calls in worker threads without sharing mutable client state:
import asyncio
from y2_client import Y2Client
async def main() -> None:
client = Y2Client.from_env()
news, feeds = await asyncio.gather(
asyncio.to_thread(
client.request_json,
"GET",
"/news?topics=markets&limit=10",
),
asyncio.to_thread(client.request_json, "GET", "/news/feeds"),
)
print(len(news["data"]), len(feeds["data"]))
asyncio.run(main())For a high-throughput service, replace the transport with an async HTTP client that supports connection pooling, but keep the same authentication, timeout, error, pagination, and media-type rules.
Handle Problem Details
from y2_client import Y2ApiError, Y2Client
client = Y2Client.from_env()
try:
client.request_json("GET", "/reports/not-a-report-id")
except Y2ApiError as error:
code = error.problem.get("code") if error.problem else "UNKNOWN"
print(error.status, code, error.request_id)
if error.status == 429:
print("Retry after", error.retry_after)Retry only safe reads or writes protected by an idempotency key. A 404 on tenant-owned resources
can intentionally avoid revealing whether another tenant owns the ID.
Call Intel v2 with the correct base
Intel path keys already contain /api/v2, and their OpenAPI operation overrides the top-level v1
server:
incidents = client.request_json(
"GET",
"https://api.y2.dev/api/v2/incidents",
query={"limit": 20},
)Do not prepend https://api.y2.dev/api/v1 to an Intel v2 path.
Evaluate a generated Python client
If you generate models and operation methods from OpenAPI, verify that the result:
- honors operation-level server overrides;
- represents nullable and optional fields separately;
- exposes
application/problem+jsonrather than replacing it with generic exceptions; - preserves response headers for rate limits, ETags, request IDs, and x402;
- handles
204and302without trying to decode JSON; and - provides explicit streaming APIs for NDJSON and Agent Y2 SSE.
A generated package can reduce boilerplate, but its published method names and release cadence are separate from the Y2 API contract. Compare its generated operation IDs with the current endpoint reference before adopting it.