SAM.gov API Complete Technical Guide

# Environment variables for API key management export SAM_API_KEY="your_32_character_api_key_here" export SAM_API_BASE_URL="https://api.sam.gov" # Test your API key curl -H "X-Api-Key: $SAM_API_KEY" \ "https://api.sam.gov/opportunities/v2/search?limit=1"

import requests import json import time from datetime import datetime, timedelta import logging class SAMAPIClient: def __init__(self, api_key): self.api_key = api_key self.base_url = "https://api.sam.gov" self.session = requests.Session() self.session.headers.update({ 'X-Api-Key': api_key, 'Accept': 'application/json' }) self.rate_limit_remaining = 10 # Track daily limit def search_opportunities(self, **params): """Search for opportunities with rate limiting""" if self.rate_limit_remaining <= 0: raise Exception("Daily rate limit exceeded (10 requests)") endpoint = "/opportunities/v2/search" url = f"{self.base_url}{endpoint}" # Default parameters default_params = { 'limit': 10, # Small limit due to rate constraints 'offset': 0, 'postedFrom': (datetime.now() - timedelta(days=30)).strftime('%m/%d/%Y') } default_params.update(params) try: response = self.session.get(url, params=default_params, timeout=30) self.rate_limit_remaining -= 1 if response.status_code == 200: data = response.json() return self._process_opportunities(data) elif response.status_code == 429: raise Exception("Rate limit exceeded") else: response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"SAM API Error: {e}") raise def _process_opportunities(self, data): """Process the complex SAM.gov response structure""" opportunities = [] # SAM.gov returns nested structure opp_data = data.get('opportunitiesData', []) for opp in opp_data: # Extract data from nested structure processed_opp = { 'id': opp.get('noticeId'), 'title': opp.get('title'), 'agency': opp.get('fullParentPathName'), 'posted_date': opp.get('postedDate'), 'response_deadline': opp.get('responseDeadLine'), 'type': opp.get('type'), 'base_type': opp.get('baseType'), # Contact info (often incomplete) 'contact': self._extract_contact_info(opp.get('pointOfContact', [])), # Set-aside info 'set_aside': opp.get('typeOfSetAside'), # Location info 'place_of_performance': opp.get('placeOfPerformance', {}), # NAICS codes 'naics': [naics.get('code') for naics in opp.get('naicsCode', [])], # Solicitation number 'solicitation_number': opp.get('solicitationNumber'), # Award info (if available) 'award': self._extract_award_info(opp.get('award', {})), # Important: Description is NOT available via API 'description': None, # Must scrape from SAM.gov website # Raw data for debugging 'raw_data': opp } opportunities.append(processed_opp) return { 'opportunities': opportunities, 'total_records': data.get('totalRecords', 0), 'rate_limit_remaining': self.rate_limit_remaining } def _extract_contact_info(self, contacts): """Extract contact information (often incomplete)""" if not contacts: return None contact = contacts[0] # Take first contact return { 'name': contact.get('fullName'), 'email': contact.get('email'), 'phone': contact.get('phone'), 'title': contact.get('title') } def _extract_award_info(self, award_data): """Extract award information if available""" if not award_data: return None return { 'amount': award_data.get('amount'), 'date': award_data.get('date'), 'awardee': award_data.get('awardee', {}).get('name') } def get_opportunity_details(self, opportunity_id): """Get detailed information for a specific opportunity""" if self.rate_limit_remaining <= 0: raise Exception("Daily rate limit exceeded") endpoint = f"/opportunities/v2/{opportunity_id}" url = f"{self.base_url}{endpoint}" try: response = self.session.get(url, timeout=30) self.rate_limit_remaining -= 1 if response.status_code == 200: return response.json() else: response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"SAM API Error: {e}") raise # Usage example def main(): # Initialize client api_key = "your_sam_gov_api_key" client = SAMAPIClient(api_key) try: # Search for recent opportunities result = client.search_opportunities( keyword="software development", limit=5 # Keep small due to rate limits ) print(f"Found {len(result['opportunities'])} opportunities") print(f"Rate limit remaining: {result['rate_limit_remaining']}") for opp in result['opportunities']: print(f"\nTitle: {opp['title']}") print(f"Agency: {opp['agency']}") print(f"Posted: {opp['posted_date']}") print(f"Deadline: {opp['response_deadline']}") print(f"NAICS: {', '.join(opp['naics']) if opp['naics'] else 'None'}") print(f"Contact: {opp['contact']['email'] if opp['contact'] else 'None'}") print(f"Description: NOT AVAILABLE VIA API") # Major limitation except Exception as e: print(f"Error: {e}") if __name__ == "__main__": main()

const axios = require('axios'); class SAMAPIClient { constructor(apiKey) { this.apiKey = apiKey; this.baseURL = 'https://api.sam.gov'; this.rateLimitRemaining = 10; // Track daily limit this.client = axios.create({ baseURL: this.baseURL, headers: { 'X-Api-Key': apiKey, 'Accept': 'application/json' }, timeout: 30000 }); } async searchOpportunities(params = {}) { if (this.rateLimitRemaining <= 0) { throw new Error('Daily rate limit exceeded (10 requests)'); } const defaultParams = { limit: 10, offset: 0, postedFrom: this._getDateString(30) // Last 30 days }; const searchParams = { ...defaultParams, ...params }; try { const response = await this.client.get('/opportunities/v2/search', { params: searchParams }); this.rateLimitRemaining--; return this._processOpportunities(response.data); } catch (error) { if (error.response?.status === 429) { throw new Error('Rate limit exceeded'); } throw new Error(`SAM API Error: ${error.message}`); } } _processOpportunities(data) { const opportunities = data.opportunitiesData?.map(opp => ({ id: opp.noticeId, title: opp.title, agency: opp.fullParentPathName, postedDate: opp.postedDate, responseDeadline: opp.responseDeadLine, type: opp.type, contact: this._extractContact(opp.pointOfContact), setAside: opp.typeOfSetAside, naics: opp.naicsCode?.map(n => n.code) || [], solicitationNumber: opp.solicitationNumber, // Description NOT available via API description: null, rawData: opp })) || []; return { opportunities, totalRecords: data.totalRecords || 0, rateLimitRemaining: this.rateLimitRemaining }; } _extractContact(contacts) { if (!contacts || contacts.length === 0) return null; const contact = contacts[0]; return { name: contact.fullName, email: contact.email, phone: contact.phone, title: contact.title }; } _getDateString(daysAgo) { const date = new Date(); date.setDate(date.getDate() - daysAgo); return date.toLocaleDateString('en-US'); } } // Usage example async function main() { const client = new SAMAPIClient('your_sam_gov_api_key'); try { const result = await client.searchOpportunities({ keyword: 'software development', limit: 5 }); console.log(`Found ${result.opportunities.length} opportunities`); console.log(`Rate limit remaining: ${result.rateLimitRemaining}`); result.opportunities.forEach(opp => { console.log(`\nTitle: ${opp.title}`); console.log(`Agency: ${opp.agency}`); console.log(`Posted: ${opp.postedDate}`); console.log(`Contact: ${opp.contact?.email || 'None'}`); console.log(`Description: NOT AVAILABLE VIA API`); }); } catch (error) { console.error(`Error: ${error.message}`); } } main();

# Python error handling example import time from requests.exceptions import RequestException, HTTPError, Timeout def robust_sam_request(client, endpoint, params=None, max_retries=3): """Make SAM API request with comprehensive error handling""" for attempt in range(max_retries): try: if client.rate_limit_remaining <= 0: raise Exception("Rate limit exhausted - wait until tomorrow") response = client.session.get( f"{client.base_url}{endpoint}", params=params, timeout=30 ) if response.status_code == 200: client.rate_limit_remaining -= 1 return response.json() elif response.status_code == 401: raise Exception("Invalid API key - check SAM.gov account") elif response.status_code == 403: raise Exception("Access denied - request Data Entry role") elif response.status_code == 429: raise Exception("Rate limit exceeded - daily limit is 10 requests") elif response.status_code >= 500: if attempt < max_retries - 1: wait_time = 2 ** attempt # Exponential backoff print(f"Server error, retrying in {wait_time} seconds...") time.sleep(wait_time) continue else: raise Exception("SAM.gov server error - try again later") else: response.raise_for_status() except Timeout: if attempt < max_retries - 1: print("Request timeout, retrying...") time.sleep(2) continue else: raise Exception("Request timeout - SAM.gov may be slow") except RequestException as e: raise Exception(f"Network error: {e}") raise Exception("Max retries exceeded")

# Production rate limiting strategy import redis from datetime import datetime, timedelta class ProductionSAMClient: def __init__(self, api_key, redis_host='localhost'): self.api_key = api_key self.redis = redis.Redis(host=redis_host, decode_responses=True) self.rate_limit_key = f"sam_api_calls_{datetime.now().strftime('%Y-%m-%d')}" def can_make_request(self): """Check if we can make another API call today""" daily_calls = int(self.redis.get(self.rate_limit_key) or 0) return daily_calls < 10 # Adjust for your plan def record_api_call(self): """Record that we made an API call""" pipe = self.redis.pipeline() pipe.incr(self.rate_limit_key) pipe.expire(self.rate_limit_key, 86400) # 24 hour TTL pipe.execute() def get_cached_result(self, cache_key): """Get cached result to avoid unnecessary API calls""" cached = self.redis.get(f"sam_cache_{cache_key}") return json.loads(cached) if cached else None def cache_result(self, cache_key, data, ttl=3600): """Cache result for 1 hour to reduce API usage""" self.redis.setex( f"sam_cache_{cache_key}", ttl, json.dumps(data) )

# Web scraping for missing descriptions from selenium import webdriver from bs4 import BeautifulSoup import time def scrape_opportunity_description(opportunity_id): """Scrape contract description from SAM.gov website""" # This is necessary because API doesn't provide descriptions try: driver = webdriver.Chrome() # Requires ChromeDriver url = f"https://sam.gov/opp/{opportunity_id}/view" driver.get(url) # Wait for page load time.sleep(3) # Find description section soup = BeautifulSoup(driver.page_source, 'html.parser') description_element = soup.find('div', class_='description-text') if description_element: return description_element.get_text(strip=True) else: return "Description not found" except Exception as e: print(f"Scraping error: {e}") return None finally: if driver: driver.quit() # This adds significant complexity and maintenance overhead

# GovCon API - Same functionality, 95% less code import requests def get_complete_opportunities(): """Get complete opportunity data in one simple request""" headers = {'Authorization': 'Bearer your_govcon_api_key'} response = requests.get( 'https://govconapi.com/api/v1/opportunities/search', headers=headers, params={ 'naics': '541330', 'state': 'CA', 'limit': 100 # 10x more data per request } ) if response.status_code == 200: data = response.json() for opp in data['data']: print(f"Title: {opp['title']}") print(f"Agency: {opp['agency']}") print(f"Posted: {opp['posted_date']}") print(f"Description: {opp['description_text'][:100]}...") # Included! print(f"Contact: {opp['contact_email']}") # Complete contact info print(f"Award Amount: ${opp['award_amount'] or 'TBD'}") # Integrated awards print(f"Winner: {opp['awardee_name'] or 'TBD'}") # Winner information print("---") return data else: raise Exception(f"API Error: {response.status_code}") # This is all you need - no registration, scraping, or complex parsing opportunities = get_complete_opportunities() print(f"Found {len(opportunities['data'])} opportunities")