fedpulse 1.0.5

Python client for the FedPulse API — federal opportunities, exclusions, entity intelligence.

FedPulse Python SDK

The official Python SDK for the FedPulse API — federal contract opportunities, exclusions, entity intelligence, and market analytics.

Requirements

• Python 3.10+
• httpx ≥ 0.27
• pydantic ≥ 2.0

Installation

pip install fedpulse

Quick Start

Synchronous

from fedpulse import FedPulse

client = FedPulse(api_key="fp_live_...")

# Search opportunities
result = client.opportunities.list(
    naics="541512",
    set_aside="SBA",
    limit=25,
)

for opp in result.data:
    print(opp.notice_id, opp.title, opp.award_amount)

print(f"Total: {result.pagination.total}")
print(f"Remaining requests: {result.rate_limit.remaining}")

Asynchronous

import asyncio
from fedpulse import AsyncFedPulse

async def main():
    async with AsyncFedPulse(api_key="fp_live_...") as client:
        result = await client.opportunities.list(naics="541512")
        for opp in result.data:
            print(opp.title)

asyncio.run(main())

Resources

Opportunities

# List with filters
result = client.opportunities.list(
    naics="541512",
    set_aside="SBA",
    department="Department of Defense",
    posted_after="2025-01-01",
    limit=25,
)

# Single opportunity
opp = client.opportunities.get("ABC123XYZ")

# Usage stats
stats = client.opportunities.stats()

# Auto-paginate (yields ParsedResponse[list[Opportunity]] pages)
for page in client.opportunities.paginate(naics="541512"):
    for opp in page.data:
        print(opp.title)

Exclusions

# List excluded entities
result = client.exclusions.list(agency="GSA")

# Check compliance for multiple entities
report = client.exclusions.check([
    {"uei": "ABCDE12345XY"},
    {"name": "Acme Corp"},
    {"cage": "1ABC2"},
])
for result in report.data:
    if result.excluded:
        print(f"{result.input.uei or result.input.name} is EXCLUDED")
        print(f"  Agency: {result.exclusion.agency}")
        print(f"  Confidence: {result.match_confidence}")

Entity Intelligence

# 360° entity profile
profile = client.intelligence.entity_profile("ABCDE12345XY")
print(profile.data.total_award_amount)
print(profile.data.agency_relationships)
print(profile.data.risk_indicators)

# Market analysis
market = client.intelligence.market_analysis("541512")
print(market.data.total_market_size)
print(market.data.top_awardees)

# Compliance package
compliance = client.intelligence.compliance_check("ABCDE12345XY")
print(compliance.data.risk_level)   # "low" | "medium" | "high"
print(compliance.data.risk_score)   # 0.0 – 1.0

Entities

# Search registered entities
result = client.entities.list(
    naics="541512",
    certification="8a",
    state="VA",
)

for entity in result.data:
    print(entity.uei, entity.legal_business_name)
    print(entity.is_8a, entity.is_wosb)     # convenience properties
    print(entity.naics_code_list)             # list[str]

# Single entity
entity = client.entities.get("ABCDE12345XY")

# Opportunities for an entity
opps = client.entities.opportunities("ABCDE12345XY")

# Exclusion check for an entity
check = client.entities.exclusion_check("ABCDE12345XY")

Analytics

# Usage summary
summary = client.analytics.summary()

# Endpoint breakdown
endpoints = client.analytics.endpoints(range="30d")

# Request logs
logs = client.analytics.logs(limit=100)

# Paginate logs
for page in client.analytics.log_pages(limit=100):
    for log in page.data.items:
        print(log)

Error Handling

from fedpulse import (
    FedPulse,
    AuthenticationError,
    PermissionError,
    RateLimitError,
    NotFoundError,
    ValidationError,
    NetworkError,
    TimeoutError,
)

client = FedPulse(api_key="fp_live_...")

try:
    result = client.opportunities.list(naics="541512")
except AuthenticationError as exc:
    # exc.code == "UNAUTHORIZED"
    print(f"Invalid API key: {exc.message}")
except PermissionError as exc:
    # exc.code == "FORBIDDEN"
    print(f"Access denied: {exc.message}")
except RateLimitError as exc:
    # exc.retry_after is seconds to wait (int | None)
    print(f"Rate limited. Retry after {exc.retry_after}s")
except NotFoundError:
    print("Resource not found")
except ValidationError as exc:
    print(f"Bad request: {exc.details}")
except NetworkError as exc:
    print(f"Network failure: {exc.message}")
except TimeoutError:
    print("Request timed out")

Pydantic Models

All responses are fully typed Pydantic v2 models. Models use snake_case in Python (alias from the API's camelCase):

entity = client.entities.get("ABCDE12345XY")
print(entity.legal_business_name)   # camelCase alias: legalBusinessName
print(entity.is_8a)                 # convenience property
print(entity.is_wosb)               # convenience property
print(entity.naics_code_list)       # list[str] derived from naics_codes

opp = client.opportunities.get("NOTICE123")
print(opp.notice_id)
print(opp.award_amount)             # str | float | int | None
print(opp.ai_tags)                  # list[str]

Pagination

paginate() is a generator that yields pages (ParsedResponse[list[T]]), not individual items:

# Sync
for page in client.opportunities.paginate(naics="541512", set_aside="SBA"):
    for opp in page.data:
        process(opp)
    # Rate limit info per page
    print(page.rate_limit.remaining)

# Async
async for page in client.opportunities.paginate(naics="541512"):
    for opp in page.data:
        await process(opp)

Webhook Verification

from fedpulse import FedPulse

# As a static method on the client
payload = FedPulse.verify_webhook(
    raw_body=request.body,           # bytes
    signature_header=request.headers["X-FedPulse-Signature"],
    timestamp_header=request.headers["X-FedPulse-Timestamp"],
    secret="whsec_...",
)
print(payload["event"], payload["data"])

# Or use the standalone function
from fedpulse import verify_webhook
payload = verify_webhook(raw_body, sig_header, ts_header, secret)

By default, webhooks older than 300 seconds are rejected. Pass max_age_seconds=600 to override.

Context Manager Support

# Sync
with FedPulse(api_key="fp_live_...") as client:
    result = client.opportunities.list()

# Async
async with AsyncFedPulse(api_key="fp_live_...") as client:
    result = await client.opportunities.list()

Configuration

client = FedPulse(
    api_key="fp_live_...",
    base_url="https://api.fedpulse.dev",  # default
    timeout=30.0,                          # seconds, default 30
    max_retries=3,                         # default 3
    cache_size=256,                        # LRU cache size, default 256
    cache_ttl_seconds=60,                  # default 60s
)

Community

• 💬 Discord — Ask questions, share what you build, get help in #help
• 🔧 Dashboard — Manage API keys, view logs and usage
• 📖 Docs — Full API reference
• 🐛 Bug reports — GitHub Issues for the API

---

License

MIT © FedPulse