Python SDK¶
The carver-feeds-sdk provides programmatic access to Carver Horizon's regulatory feed data in Python. It is the fastest way to query, filter, and analyze regulatory content without building raw API calls.
Installation¶
Contact Carver Agents for install instructions.
Configuration¶
Create a .env file in your project directory:
Generate your API key from Admin Dashboard → Settings → API Keys.
Three Components¶
The SDK provides three levels of abstraction — use the one that fits your task:
| Component | Use when | Returns |
|---|---|---|
CarverFeedsAPIClient |
You need direct API control or raw responses | Python dicts |
FeedsDataManager |
You want simple DataFrame operations | pandas DataFrames |
EntryQueryEngine |
You need complex multi-filter queries | DataFrames, CSV, JSON |
API Client¶
Low-level HTTP client. Use this when you want raw API responses:
from dotenv import load_dotenv
from carver_feeds import get_client
load_dotenv()
client = get_client()
# List all regulatory categories
categories = client.list_categories()
print(f"Found {len(categories)} categories")
# List topics, optionally filtered by category
topics = client.list_topics(category_id="cat-uuid-123")
# Get entries for a specific topic
entries = client.get_topic_entries(topic_id="topic-uuid")
print(f"Found {len(entries)} entries")
Data Manager¶
Converts API responses to pandas DataFrames:
from dotenv import load_dotenv
from carver_feeds import create_data_manager
load_dotenv()
dm = create_data_manager()
# Get categories as a DataFrame
categories_df = dm.get_categories_df()
print(categories_df[['id', 'name', 'topic_count']].head())
# Get entries for a topic
entries_df = dm.get_topic_entries_df(topic_id="topic-uuid")
print(f"Found {len(entries_df)} entries")
# Fetch entry content from S3 (opt-in — expensive for large result sets)
entries_df = dm.get_topic_entries_df(topic_id="topic-uuid", fetch_content=True)
Query Engine¶
High-level fluent interface for building complex queries with method chaining. See Query Engine for full documentation.
from dotenv import load_dotenv
from carver_feeds import create_query_engine
from datetime import datetime
load_dotenv()
qe = create_query_engine()
results = qe \
.filter_by_category(category_name="Finance") \
.filter_by_date(start_date=datetime(2024, 1, 1)) \
.search_entries("capital requirements") \
.to_dataframe()
print(f"Found {len(results)} matching entries")
Data Model¶
Regulatory data is organized in a four-level hierarchy:
Category (e.g., "Finance", "Medical Devices", "Data Protection")
└─ Topic (regulatory body, e.g., "SEC", "SEBI", "RBI")
└─ Entry (individual article or regulatory notice)
└─ Annotation (AI-generated insights and scores)
Key fields by entity:
| Field | Type | Description |
|---|---|---|
id |
UUID | Unique identifier |
name |
string | Display name |
slug |
string | URL-friendly name |
is_active |
bool | Whether category is active |
topic_count |
int | Number of topics in category |
| Field | Type | Description |
|---|---|---|
id |
UUID | Unique identifier |
name |
string | Display name |
is_active |
bool | Whether topic is active |
acronym |
string | Short code (requires details=True) |
jurisdiction_code |
string | Country/jurisdiction (requires details=True) |
sectors |
list | Industry sectors covered (requires details=True) |
| Field | Type | Description |
|---|---|---|
id |
UUID | Unique identifier |
title |
string | Article title |
link |
string | Original source URL |
published_date |
datetime | When the article was published |
is_active |
bool | Whether entry is active |
extracted_metadata.topic_id |
UUID | Parent topic |
extracted_metadata.status |
string | Processing status |
| Field | Type | Description |
|---|---|---|
scores.impact |
float | Impact score (0–1) |
scores.urgency |
float | Urgency score (0–1) |
scores.relevance |
float | Relevance score (0–1) |
classification.update_type |
string | Type of regulatory change |
metadata.impact_summary |
string | AI-generated summary |
metadata.critical_dates |
list | Key compliance deadlines |
Error Handling¶
from carver_feeds import get_client
from carver_feeds.exceptions import AuthenticationError, RateLimitError, CarverAPIError
client = get_client()
try:
entries = client.get_topic_entries(topic_id="topic-uuid")
except AuthenticationError:
print("Invalid API key — check your CARVER_API_KEY")
except RateLimitError:
print("Rate limit hit — the SDK retries automatically, but max retries exceeded")
except CarverAPIError as e:
print(f"API error: {e}")
The SDK automatically retries on rate limits (HTTP 429) and server errors (HTTP 500) with exponential backoff. Authentication errors (401/403) and not-found errors (404) are not retried.
Next Steps¶
- Query Engine — Complex multi-filter queries with method chaining
- Annotations — Access AI-generated insights and scores
- API Reference — Raw endpoint documentation