feat: market-brief + agent-subscription methods (sync+async, v1.9.0)

README.md

@@ -426,6 +426,75 @@ result = client.data_sources.test("123")
 print(f"Test status: {result['status']}")
 ```
 
+### Market Brief (New in v1.9.0)
+
+Get a multi-commodity structured summary (latest price, 24h change, and a
+1-month forecast per commodity) in a single request. Pass `narrative=True` to
+also receive a natural-language summary.
+
+```python
+brief = client.market_brief(["BRENT_CRUDE_USD", "WTI_USD"], narrative=True)
+
+print(f"As of: {brief.as_of}")
+for c in brief.commodities:
+    print(f"{c.code}: ${c.price} ({c.change_24h_pct:+.2f}% 24h)")
+    if c.forecast_1m:
+        print(f"  1m forecast: {c.forecast_1m.point} "
+              f"[{c.forecast_1m.low}–{c.forecast_1m.high}] ({c.forecast_1m.confidence})")
+
+if brief.narrative:
+    print(brief.narrative)
+```
+
+### Agent Subscriptions (New in v1.9.0)
+
+Create persistent "watches" that periodically evaluate commodities and emit
+events your agent can poll for. The `interval` accepts a friendly string
+(`"5m"`, `"1h"`, `"daily"`) or raw seconds.
+
+```python
+# Create a subscription
+sub = client.subscriptions.create(
+    codes=["BRENT_CRUDE_USD"],
+    interval="5m",            # also accepts "1h", "daily", or 300
+    name="Brent watch",
+)
+print(sub.id, sub.interval_seconds)  # -> 300
+
+# List subscriptions
+for s in client.subscriptions.list():
+    print(s.name, s.codes, s.status)
+
+# Poll for events using a cursor
+page = client.subscriptions.events(since=0)
+for event in page:
+    print(event.type, event.code)
+# Persist page.cursor and pass it as `since` on the next poll
+next_page = client.subscriptions.events(since=page.cursor)
+
+# Delete a subscription
+client.subscriptions.delete(sub.id)
+```
+
+Attribution headers are sent automatically (`X-OPA-Source` defaults to
+`sdk-python`). MCP tools can override them:
+
+```python
+client.subscriptions.create(
+    ["WTI_USD"], interval="1h", source="mcp", tool="claude-desktop"
+)
+```
+
+All of the above is mirrored on the async client:
+
+```python
+async with AsyncOilPriceAPI() as client:
+    brief = await client.market_brief(["BRENT_CRUDE_USD"])
+    sub = await client.subscriptions.create(["WTI_USD"], interval="daily")
+    page = await client.subscriptions.events(since=0)
+    await client.subscriptions.delete(sub.id)
+```
+
 ## 📊 Features
 
 - ✅ **Simple API** - Intuitive methods for all endpoints
@@ -444,6 +513,8 @@ print(f"Test status: {result['status']}")
 - ✅ **Energy Intelligence** - EIA data, OPEC production, drilling productivity
 - ✅ **Data Quality** - Real-time quality monitoring and reporting
 - ✅ **Data Sources** - Connector management with health checks and logging
+- ✅ **Market Brief** - Multi-commodity structured + narrative summary in one call 🧠
+- ✅ **Agent Subscriptions** - Persistent watches + event polling for AI agents 🤖
 - ✅ **Async Support** - High-performance async client
 - ✅ **WebSocket Streaming** - Real-time price stream via ActionCable (Professional+)
 - ✅ **Smart Caching** - Reduce API calls automatically

oilpriceapi/__init__.py

@@ -28,9 +28,15 @@
     DieselPrice,
     DieselStation,
     DieselStationsResponse,
+    MarketBrief,
+    MarketBriefCommodity,
+    MarketBriefForecast,
     PriceAlert,
+    Subscription,
+    SubscriptionEvent,
     WebhookTestResponse,
 )
+from oilpriceapi.resources.subscriptions import SubscriptionEventsPage
 from oilpriceapi.streaming import (
     PriceStream,
     PriceUpdate,
@@ -56,6 +62,12 @@
     "PriceAlert",
     "WebhookTestResponse",
     "DataConnectorPrice",
+    "MarketBrief",
+    "MarketBriefCommodity",
+    "MarketBriefForecast",
+    "Subscription",
+    "SubscriptionEvent",
+    "SubscriptionEventsPage",
     "PriceStream",
     "StreamUpdate",
     "PriceUpdate",

oilpriceapi/_subscriptions_common.py

@@ -0,0 +1,111 @@
+"""
+Shared helpers for the agent-subscriptions + market-brief features (#3245).
+
+Keeps the sync and async resources behaving identically: friendly interval
+parsing, attribution header construction, and response unwrapping.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Union
+
+# Default attribution source stamped on subscriptions created via this SDK.
+DEFAULT_SOURCE = "sdk-python"
+
+# Friendly interval aliases → seconds. Anything else falls through to the
+# numeric parser below (e.g. "30s", "15m", "2h", "1d", or a raw int).
+_INTERVAL_ALIASES: Dict[str, int] = {
+    "1m": 60,
+    "5m": 300,
+    "10m": 600,
+    "15m": 900,
+    "30m": 1800,
+    "1h": 3600,
+    "hourly": 3600,
+    "6h": 21600,
+    "12h": 43200,
+    "1d": 86400,
+    "daily": 86400,
+}
+
+_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
+
+
+def normalize_interval(interval: Union[str, int]) -> int:
+    """Convert a friendly interval into ``interval_seconds``.
+
+    Accepts an int (returned as-is), a named alias ("5m", "1h", "daily"), or a
+    ``<number><unit>`` string where unit is one of s/m/h/d (e.g. "30s", "2h").
+
+    Raises:
+        ValueError: If the interval cannot be parsed or is non-positive.
+    """
+    if isinstance(interval, bool):  # bool is an int subclass; reject explicitly
+        raise ValueError(f"Invalid interval: {interval!r}")
+    if isinstance(interval, int):
+        if interval <= 0:
+            raise ValueError(f"interval_seconds must be positive, got {interval}")
+        return interval
+
+    if not isinstance(interval, str):
+        raise ValueError(f"Invalid interval type: {type(interval).__name__}")
+
+    key = interval.strip().lower()
+    if key in _INTERVAL_ALIASES:
+        return _INTERVAL_ALIASES[key]
+
+    # Plain integer string e.g. "300".
+    if key.isdigit():
+        seconds = int(key)
+        if seconds <= 0:
+            raise ValueError(f"interval_seconds must be positive, got {seconds}")
+        return seconds
+
+    # <number><unit> form e.g. "45s", "2h".
+    if len(key) >= 2 and key[-1] in _UNIT_SECONDS and key[:-1].isdigit():
+        seconds = int(key[:-1]) * _UNIT_SECONDS[key[-1]]
+        if seconds <= 0:
+            raise ValueError(f"interval_seconds must be positive, got {seconds}")
+        return seconds
+
+    raise ValueError(
+        f"Unrecognized interval {interval!r}. Use seconds (int), a named alias "
+        f"('5m', '1h', 'daily'), or '<n><unit>' where unit is s/m/h/d."
+    )
+
+
+def build_attribution_headers(
+    source: Optional[str] = None,
+    tool: Optional[str] = None,
+) -> Dict[str, str]:
+    """Build the X-OPA-Source / X-OPA-Tool attribution headers.
+
+    Defaults ``source`` to ``sdk-python``; omits the tool header when unset.
+    """
+    headers: Dict[str, str] = {"X-OPA-Source": source or DEFAULT_SOURCE}
+    if tool:
+        headers["X-OPA-Tool"] = tool
+    return headers
+
+
+def build_create_body(
+    codes: List[str],
+    interval: Union[str, int],
+    name: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Build the POST /v1/subscriptions request body from friendly args."""
+    body: Dict[str, Any] = {
+        "codes": list(codes),
+        "interval_seconds": normalize_interval(interval),
+    }
+    if name is not None:
+        body["name"] = name
+    return body
+
+
+def unwrap_data(response: Any) -> Dict[str, Any]:
+    """Return the ``data`` object from a ``{status, data}`` envelope."""
+    if isinstance(response, dict) and "data" in response:
+        data = response["data"]
+        return data if isinstance(data, dict) else {}
+    return response if isinstance(response, dict) else {}

oilpriceapi/async_client.py

@@ -17,6 +17,7 @@
 
 logger = logging.getLogger(__name__)
 
+from ._subscriptions_common import unwrap_data
 from .async_resources import (
     AsyncAlertsResource,
     AsyncAnalyticsResource,
@@ -31,6 +32,7 @@
     AsyncFuturesResource,
     AsyncRigCountsResource,
     AsyncStorageResource,
+    AsyncSubscriptionsResource,
     AsyncWebhooksResource,
 )
 from .exceptions import (
@@ -42,7 +44,7 @@
     ServerError,
     TimeoutError,
 )
-from .models import HistoricalPrice, HistoricalResponse, Price
+from .models import HistoricalPrice, HistoricalResponse, MarketBrief, Price
 from .retry import RetryStrategy
 
 
@@ -148,6 +150,8 @@ def __init__(
         self.ei = AsyncEnergyIntelligenceResource(self)
         self.webhooks = AsyncWebhooksResource(self)
         self.data_sources = AsyncDataSourcesResource(self)
+        # Agent watch subscriptions + event polling (#3245 Phase 2).
+        self.subscriptions = AsyncSubscriptionsResource(self)
 
         # Real-time WebSocket streaming namespace (requires the [stream] extra).
         # Lazily imports `websockets` only when a stream is actually opened.
@@ -342,6 +346,34 @@ def _parse_rate_limit_reset(self, headers: httpx.Headers) -> Optional[datetime]:
                     pass
         return None
 
+    async def market_brief(
+        self,
+        codes: List[str],
+        narrative: bool = False,
+    ) -> MarketBrief:
+        """Get a multi-commodity structured (+ optional narrative) market brief.
+
+        Composes existing price/forecast data for the given commodity codes into
+        a single structured summary (#3245 Phase 1a). Counts as one request.
+
+        Args:
+            codes: Commodity codes to include (e.g. ["BRENT_CRUDE_USD", "WTI"]).
+            narrative: When True, request a natural-language narrative as well.
+
+        Returns:
+            A MarketBrief model.
+        """
+        params: Dict[str, Any] = {"codes": ",".join(codes)}
+        if narrative:
+            params["narrative"] = "true"
+
+        response = await self.request(
+            method="GET",
+            path="/v1/market-brief",
+            params=params,
+        )
+        return MarketBrief(**unwrap_data(response))
+
     async def close(self):
         """Close the HTTP client and flush telemetry."""
         self._telemetry.close()

oilpriceapi/async_resources.py

@@ -3,10 +3,16 @@
 from datetime import date, datetime
 from typing import Any, Dict, List, Optional, Union
 
+from ._subscriptions_common import (
+    build_attribution_headers,
+    build_create_body,
+    unwrap_data,
+)
 from .exceptions import ValidationError
-from .models import DieselPrice, DieselStationsResponse, PriceAlert
+from .models import DieselPrice, DieselStationsResponse, PriceAlert, Subscription, SubscriptionEvent
 from .resource_validators import VALID_OPERATORS, format_date
 from .resources._futures_slug import normalize_futures_slug
+from .resources.subscriptions import SubscriptionEventsPage
 
 
 class AsyncDieselResource:
@@ -1403,3 +1409,83 @@ async def rotate_credentials(self, source_id: str, new_credentials: Dict[str, An
         if "data" in response:
             return response["data"]
         return response
+
+
+class AsyncSubscriptionsResource:
+    """Async resource for agent-subscription CRUD and event polling (#3245)."""
+
+    def __init__(self, client: Any) -> None:
+        self.client = client
+
+    async def list(self) -> List[Subscription]:
+        """List all subscriptions for the authenticated user."""
+        response = await self.client.request(method="GET", path="/v1/subscriptions")
+        data = unwrap_data(response)
+        subs = data.get("subscriptions", [])
+        return [Subscription(**s) for s in subs]
+
+    async def create(
+        self,
+        codes: List[str],
+        interval: Union[str, int],
+        name: Optional[str] = None,
+        source: Optional[str] = None,
+        tool: Optional[str] = None,
+    ) -> Subscription:
+        """Create a new subscription (watch).
+
+        Args:
+            codes: Commodity codes to watch (e.g. ["BRENT_CRUDE_USD"]).
+            interval: Friendly interval ("5m", "1h", "daily") or seconds (int).
+            name: Optional human-friendly name.
+            source: Attribution source header (defaults to "sdk-python").
+            tool: Optional attribution tool name header.
+        """
+        body = build_create_body(codes, interval, name=name)
+        headers = build_attribution_headers(source=source, tool=tool)
+        response = await self.client.request(
+            method="POST",
+            path="/v1/subscriptions",
+            json_data=body,
+            headers=headers,
+        )
+        data = unwrap_data(response)
+        sub = data.get("subscription", data)
+        return Subscription(**sub)
+
+    async def delete(self, subscription_id: str) -> bool:
+        """Delete a subscription. Returns True on success."""
+        await self.client.request(
+            method="DELETE",
+            path=f"/v1/subscriptions/{subscription_id}",
+        )
+        return True
+
+    async def events(
+        self,
+        since: Optional[int] = None,
+        limit: Optional[int] = None,
+        watch_id: Optional[str] = None,
+    ) -> SubscriptionEventsPage:
+        """Poll for subscription events newer than a cursor.
+
+        Returns a SubscriptionEventsPage with events, cursor, and has_more.
+        """
+        params: Dict[str, Any] = {}
+        if since is not None:
+            params["since"] = since
+        if limit is not None:
+            params["limit"] = limit
+        if watch_id is not None:
+            params["watch_id"] = watch_id
+
+        response = await self.client.request(
+            method="GET",
+            path="/v1/subscriptions/events",
+            params=params,
+        )
+        data = unwrap_data(response)
+        events = [SubscriptionEvent(**e) for e in data.get("events", [])]
+        cursor = data.get("cursor")
+        has_more = bool(data.get("has_more", False))
+        return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more)