AI-Powered Job Referral Platform
Build a FastAPI backend that scrapes jobs and matches them with Claude AI.
Introduction
30 Second Summary
Job searching is exhausting. You scroll through hundreds of listings on job boards, but the best opportunities often come through personal referrals from people you already know.
In this project, you will build ReferralRadar, a Python API that scrapes real job listings from company career pages, uses AI to score how well each job matches your profile, and maps which of your personal connections could refer you.
What You'll Build
A FastAPI backend that combines web scraping, AI-powered job matching, and referral mapping into a single job intelligence platform.
By the end of this project, you'll have:
- A FastAPI backend with auto-generated interactive Swagger documentation you can use to test every endpoint.
- A Greenhouse API scraper that fetches real job listings directly from company career pages.
- An AI matching engine using Claude that scores jobs 0-100 against your skills and experience.
- A referral mapping system that cross-references your personal network with companies that have matching jobs.
- Secret Mission: Build a live HTML dashboard showing your top matched jobs with referral-ready connections highlighted.
Are there any prerequisites?
This project assumes intermediate Python experience. You should be comfortable with functions, dictionaries, and working with JSON data.
Familiarity with APIs (making HTTP requests and reading JSON responses) will help you move through the steps faster.
Set Up Your Python Environment
Your Python environment is ready with FastAPI, anthropic, and httpx installed. Before writing any application logic, you need to connect your project to Claude's API and create the file structure that each module will live in.
ReferralRadar is split across five Python files, each handling one responsibility. This step sets up that skeleton so every future step has a clear home for its code.
In this step, get ready to:
- Set your Anthropic API key as an environment variable.
- Create the project file structure with all module files.
- Write the FastAPI app skeleton with imports and initialization.
Set your Anthropic API key
The anthropic Python SDK automatically reads the ANTHROPIC_API_KEY environment variable when you create a client. Setting it once in your terminal session means you never have to hardcode your key in source files.
- In your Cursor terminal (with the virtual environment still active), set the environment variable by running this command:
macOS
export ANTHROPIC_API_KEY=your-api-key-hereWindows
$env:ANTHROPIC_API_KEY = "your-api-key-here"- Replace your-api-key-here with the actual API key you obtained from your Anthropic account.
Why an environment variable?
Environment variables keep secrets out of your code. If you ever push this project to GitHub, your key stays safe because it only lives in your terminal session.
Key not being read later?
This variable only lasts for your current terminal session. If you close and reopen the terminal, you need to run the export command again.
Make sure there are no spaces around the = sign. The format is ANTHROPIC_API_KEY=sk-ant-... with no gaps.
help me debug my Anthropic API key setup
Create the project file structure
ReferralRadar uses five Python files, each responsible for one part of the system. Creating them now means your imports will resolve correctly when you write the app skeleton.
- database.py handles SQLite connections and schema creation.
- models.py defines Pydantic models for API validation.
- scraper.py fetches jobs from company career pages.
- matcher.py sends job descriptions to Claude for scoring.
- main.py ties everything together as the FastAPI application.
- Create all five files at once by running this command:
macOS
touch database.py models.py scraper.py matcher.py main.pyWindows
New-Item database.py, models.py, scraper.py, matcher.py, main.py- Confirm the files exist by running:
lsYou should see database.py, models.py, scraper.py, matcher.py, and main.py listed in the output.
Files not appearing?
Make sure your terminal is in the referralradar directory. Run pwd (macOS) or Get-Location (Windows) to check your current path.
help me figure out why my files aren't appearing in the project folder
Write the FastAPI app skeleton in main.py
The main.py file is the entry point for your API. It imports functions from every other module and wires them into a FastAPI application.
- Open main.py in Cursor by clicking it in the Explorer sidebar.
- Add the import block at the top of the file by pasting this code:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profileWhat do these imports do?
The first group pulls in FastAPI's core tools. UploadFile handles CSV uploads, HTTPException returns error responses, and BaseModel from Pydantic validates incoming data.
The second group imports from your own modules. These won't work yet because the files are empty. You'll fill them in over the next steps.
- Below the imports, add the app initialization code:
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize the database when the app starts
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)What does lifespan do?
The lifespan function runs setup code when your server starts. Here it calls init_db() to create database tables before the first request arrives.
The yield separates startup logic from shutdown logic. Everything before yield runs on startup. You could add cleanup code after it for shutdown.
- Save main.py.
You won't be able to run the server yet because the other modules are still empty. That's expected. The next step fills in the database and models.
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and move on to the next step.
ⓧ I'd like to double check the full code
Here's the complete main.py as it should look at the end of this step:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profile
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize the database when the app starts
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)The other four files (database.py, models.py, scraper.py, matcher.py) should exist but remain empty for now.
Your project skeleton is in place. Next up, you'll fill in the database schema and Pydantic models that give ReferralRadar its data backbone.
Design the Database and Models
Before building any API endpoints, you need a solid data foundation. ReferralRadar tracks multiple types of information: your profile, your network connections, companies you're monitoring, scraped job listings, and AI-generated match scores.
In this step, you'll create the SQLite database schema that stores all this data and the Pydantic models that validate every request and response flowing through your API.
In this step, get ready to:
- Create a SQLite database schema with five tables for profiles, connections, companies, jobs, and matches.
- Define Pydantic models for API request and response validation.
- Initialize the database and verify the file is created.
Create the database connection helper
Every part of ReferralRadar needs to read and write data. A connection helper gives all your modules a consistent way to access the database without repeating setup logic.
- Open database.py in your editor and add the connection helper by pasting this code:
import sqlite3
# Path to the SQLite database file
DATABASE_PATH = "referralradar.db"
def get_db() -> sqlite3.Connection:
# Create and return a new database connection
db = sqlite3.connect(DATABASE_PATH)
return dbWhat does this code do?
DATABASE_PATH defines where SQLite stores the database file on disk. Every time your app needs to read or write data, it calls get_db() to open a fresh connection to that file.
SQLite stores the entire database in a single file. No separate server process needed.
Define the database schema
ReferralRadar needs five tables to track your job search data. The init_db() function creates all of them when the app starts.
- Add the init_db() function below get_db() in database.py:
def init_db():
db = get_db()
# Create all tables if they don't already exist
db.executescript("""
CREATE TABLE IF NOT EXISTS profiles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
skills TEXT NOT NULL,
target_titles TEXT NOT NULL,
preferred_locations TEXT NOT NULL,
experience_level TEXT NOT NULL,
resume_summary TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS connections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
company TEXT NOT NULL,
title TEXT NOT NULL,
email TEXT,
relationship_strength TEXT DEFAULT 'medium',
notes TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS companies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
careers_url TEXT NOT NULL,
greenhouse_board_token TEXT
);
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
company TEXT NOT NULL,
title TEXT NOT NULL,
location TEXT NOT NULL,
job_url TEXT NOT NULL,
department TEXT,
description TEXT,
date_posted TEXT,
employment_type TEXT,
source_url TEXT
);
CREATE TABLE IF NOT EXISTS matches (
profile_id INTEGER NOT NULL,
job_id INTEGER NOT NULL,
score INTEGER NOT NULL,
reasoning TEXT,
PRIMARY KEY (profile_id, job_id)
);
""")
db.commit()
db.close()What do these tables store?
- profiles stores your skills, target job titles, preferred locations, and a resume summary.
- connections holds your network contacts with their company, title, and relationship strength.
- companies tracks which companies to scrape, including their Greenhouse board token.
- jobs stores scraped job listings with title, location, department, and description.
- matches records AI-generated scores (0-100) linking a profile to a job.
The CREATE TABLE IF NOT EXISTS pattern means the app can restart safely without destroying existing data.
- Save database.py.
Seeing a syntax error in the SQL?
Make sure each CREATE TABLE statement ends with a semicolon. Also check that the triple-quoted string uses three double quotes """ on both sides.
help me fix a syntax error in my database.py SQL schema
✔️ Awesome, I've got everything!
Great. Double check you've saved database.py before moving on.
ⓧ I'd like to double check the full code
import sqlite3
DATABASE_PATH = "referralradar.db"
def get_db() -> sqlite3.Connection:
db = sqlite3.connect(DATABASE_PATH)
return db
def init_db():
db = get_db()
db.executescript("""
CREATE TABLE IF NOT EXISTS profiles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
skills TEXT NOT NULL,
target_titles TEXT NOT NULL,
preferred_locations TEXT NOT NULL,
experience_level TEXT NOT NULL,
resume_summary TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS connections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
company TEXT NOT NULL,
title TEXT NOT NULL,
email TEXT,
relationship_strength TEXT DEFAULT 'medium',
notes TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS companies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
careers_url TEXT NOT NULL,
greenhouse_board_token TEXT
);
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
company TEXT NOT NULL,
title TEXT NOT NULL,
location TEXT NOT NULL,
job_url TEXT NOT NULL,
department TEXT,
description TEXT,
date_posted TEXT,
employment_type TEXT,
source_url TEXT
);
CREATE TABLE IF NOT EXISTS matches (
profile_id INTEGER NOT NULL,
job_id INTEGER NOT NULL,
score INTEGER NOT NULL,
reasoning TEXT,
PRIMARY KEY (profile_id, job_id)
);
""")
db.commit()
db.close()Define Pydantic models for request and response validation
FastAPI uses Pydantic models to automatically validate incoming request data and structure outgoing responses. Each model defines the exact fields a request must contain or a response will return.
- Open models.py and add the profile and connection models by pasting this code:
from pydantic import BaseModel
# Profile models - what the user sends (Create) vs what the API returns (Response)
class ProfileCreate(BaseModel):
name: str
skills: list[str]
target_titles: list[str]
preferred_locations: list[str]
experience_level: str
resume_summary: str
class ProfileResponse(BaseModel):
id: int
name: str
skills: list[str]
target_titles: list[str]
preferred_locations: list[str]
experience_level: str
resume_summary: str
# Connection models - your network contacts
class ConnectionCreate(BaseModel):
name: str
company: str
title: str
email: str | None = None
relationship_strength: str = "medium"
notes: str = ""
class ConnectionResponse(BaseModel):
id: int
name: str
company: str
title: str
email: str | None
relationship_strength: str
notes: strWhy separate Create and Response models?
The Create model defines what data the user sends when creating a record. The Response model includes the auto-generated id field that the database assigns.
Fields like email: str | None = None mean the field is optional with a default of None. FastAPI uses these type hints to validate requests automatically and return helpful error messages when data is missing or malformed.
- Add the remaining models below the ConnectionResponse class in models.py:
# Company models - companies you want to track
class CompanyCreate(BaseModel):
name: str
careers_url: str
greenhouse_board_token: str | None = None
class CompanyResponse(BaseModel):
id: int
name: str
careers_url: str
greenhouse_board_token: str | None
# Job listing model - scraped from career pages
class JobResponse(BaseModel):
id: int
company: str
title: str
location: str
job_url: str
department: str | None
description: str | None
date_posted: str | None
employment_type: str | None
source_url: str | None
# Match model - AI-scored job matches
class MatchResponse(BaseModel):
job_id: int
score: int
reasoning: str
company: str
title: str
location: str
job_url: str
# Referral model - matched jobs where you have a connection
class ReferralJobResponse(BaseModel):
job_id: int
score: int
reasoning: str
company: str
title: str
location: str
job_url: str
connection_name: str
connection_title: str
connection_email: str | NoneWhat does each model represent?
- CompanyCreate and CompanyResponse handle companies you want to monitor for job openings.
- JobResponse represents a scraped job listing with all its details.
- MatchResponse pairs a job with its AI-generated match score and reasoning.
- ReferralJobResponse extends MatchResponse with the connection who could refer you at that company.
- Save models.py.
Seeing a type error on list[str] or str | None?
These type hints require Python 3.10 or higher. Run python3 --version to confirm your version. If you're on an older version, revisit Step 1 to upgrade.
help me fix a type hint error in my Pydantic models
✔️ Awesome, I've got everything!
Great. Double check you've saved models.py before moving on.
ⓧ I'd like to double check the full code
from pydantic import BaseModel
class ProfileCreate(BaseModel):
name: str
skills: list[str]
target_titles: list[str]
preferred_locations: list[str]
experience_level: str
resume_summary: str
class ProfileResponse(BaseModel):
id: int
name: str
skills: list[str]
target_titles: list[str]
preferred_locations: list[str]
experience_level: str
resume_summary: str
class ConnectionCreate(BaseModel):
name: str
company: str
title: str
email: str | None = None
relationship_strength: str = "medium"
notes: str = ""
class ConnectionResponse(BaseModel):
id: int
name: str
company: str
title: str
email: str | None
relationship_strength: str
notes: str
class CompanyCreate(BaseModel):
name: str
careers_url: str
greenhouse_board_token: str | None = None
class CompanyResponse(BaseModel):
id: int
name: str
careers_url: str
greenhouse_board_token: str | None
class JobResponse(BaseModel):
id: int
company: str
title: str
location: str
job_url: str
department: str | None
description: str | None
date_posted: str | None
employment_type: str | None
source_url: str | None
class MatchResponse(BaseModel):
job_id: int
score: int
reasoning: str
company: str
title: str
location: str
job_url: str
class ReferralJobResponse(BaseModel):
job_id: int
score: int
reasoning: str
company: str
title: str
location: str
job_url: str
connection_name: str
connection_title: str
connection_email: str | NoneInitialize the database and verify it works
Your main.py already calls init_db() in its lifespan function. You can test the database setup directly without starting the full server.
- Initialize the database by running this command in your terminal:
python -c "from database import init_db; init_db()"- Confirm the database file was created by running:
ls referralradar.db- You should see referralradar.db in the output. This means all five tables were created successfully.
Don't see referralradar.db?
Make sure you're running the command from inside your referral-radar project folder where database.py lives. Also confirm your virtual environment is active (you should see the environment name in your terminal prompt).
help me debug why my database initialization isn't creating the referralradar.db file
Your data layer is ready. Next up, you'll build the API endpoints that let you create profiles, add connections, and upload your LinkedIn network via CSV.
Build the Profile and Connections API
Your database and models are ready. Now you need a way to actually interact with them.
In this step, you will build the REST API endpoints that let users create profiles, manage their connections, and bulk-import contacts from a LinkedIn CSV export.
In this step, get ready to:
- Create profile endpoints for storing your skills and preferences.
- Build connection management endpoints for adding and listing contacts.
- Implement LinkedIn CSV parsing that extracts name, company, title, and email.
Create the profile endpoints
Before we write any endpoints, the server needs placeholder functions in scraper.py and matcher.py so the imports in main.py can resolve.
- Add a placeholder function to scraper.py by pasting this code:
import httpx
def scrape_greenhouse_jobs(board_token: str, company_name: str) -> list[dict]:
# Placeholder - we'll implement this in a later step
return []- Add a placeholder function to matcher.py by pasting this code:
import anthropic
import json
def match_jobs_for_profile(profile: dict, job_rows: list) -> list[dict]:
# Placeholder - we'll implement this in a later step
return []Why do we need placeholders?
Your main.py already imports scrape_greenhouse_jobs and match_jobs_for_profile. Python will throw an ImportError if those functions don't exist. The placeholders let the server start while we focus on the profile and connections logic first.
- Save both files.
Now let's build the first endpoint. A profile stores everything the AI matcher needs to score jobs against your background.
- In main.py, find the line that ends with lifespan=lifespan, and the closing ) beneath it. Add the following code below that closing parenthesis:
@app.post("/profile", response_model=ProfileResponse)
def create_profile(profile: ProfileCreate):
db = get_db()
# Insert profile data, storing lists as JSON strings
cursor = db.execute(
"""INSERT INTO profiles (name, skills, target_titles, preferred_locations, experience_level, resume_summary)
VALUES (?, ?, ?, ?, ?, ?)""",
(
profile.name,
json.dumps(profile.skills),
json.dumps(profile.target_titles),
json.dumps(profile.preferred_locations),
profile.experience_level,
profile.resume_summary,
),
)
db.commit()
# Return the created profile with its new database ID
return ProfileResponse(
id=cursor.lastrowid,
name=profile.name,
skills=profile.skills,
target_titles=profile.target_titles,
preferred_locations=profile.preferred_locations,
experience_level=profile.experience_level,
resume_summary=profile.resume_summary,
)What does this code do?
The @app.post("/profile") decorator registers this function as a POST endpoint at /profile. FastAPI automatically validates incoming JSON against your ProfileCreate model.
The json.dumps() calls convert Python lists (skills, target_titles, preferred_locations) into JSON strings for SQLite storage. SQLite doesn't have a native list type, so JSON serialization is the standard workaround.
- You won't see a change yet. The next function lets you retrieve the profile you just created.
- Directly below the create_profile function, add the GET endpoint:
@app.get("/profile/{profile_id}", response_model=ProfileResponse)
def get_profile(profile_id: int):
db = get_db()
# Fetch the profile row by ID
row = db.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profile not found")
# Deserialize JSON strings back into Python lists
return ProfileResponse(
id=row[0],
name=row[1],
skills=json.loads(row[2]),
target_titles=json.loads(row[3]),
preferred_locations=json.loads(row[4]),
experience_level=row[5],
resume_summary=row[6],
)What does this code do?
The {profile_id} in the route becomes a function parameter that FastAPI automatically converts to an integer. If the profile doesn't exist, the endpoint returns a 404 error.
The json.loads() calls reverse the serialization from create_profile. They convert JSON strings back into Python lists so the response matches the ProfileResponse model.
- Save main.py.
- Start the FastAPI dev server by running this command in your terminal:
fastapi dev main.py- Open your browser and go to http://127.0.0.1:8000/docs.
You should see the interactive Swagger documentation page with your two profile endpoints listed.
What is Swagger?
FastAPI automatically generates interactive API documentation at the /docs path. You can expand any endpoint, fill in sample data, and click Execute to send a real request to your running server. This removes the need for separate API testing tools.
- Test the POST /profile endpoint by clicking on it, then clicking Try it out.
- Replace the example JSON body with your own details (skills, target titles, etc.) and click Execute.
- You should see a 200 response containing your profile data with an assigned id of 1.
Seeing an import error in your terminal?
If the server fails to start with ImportError, double check that scraper.py and matcher.py both have their placeholder functions saved. The function names must match exactly: scrape_greenhouse_jobs and match_jobs_for_profile.
Help me debug my FastAPI import error
Build the connection management endpoints
Connections represent people in your network who might refer you to jobs at their companies. These endpoints let you add contacts one by one or list all stored connections.
- In main.py, add the following code below the get_profile function:
@app.post("/connections", response_model=ConnectionResponse)
def add_connection(connection: ConnectionCreate):
db = get_db()
# Insert a single connection into the database
cursor = db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(
connection.name,
connection.company,
connection.title,
connection.email,
connection.relationship_strength,
connection.notes,
),
)
db.commit()
# Use model_dump() to convert the Pydantic model to a dict for the response
return ConnectionResponse(id=cursor.lastrowid, **connection.model_dump())What does model_dump() do?
Pydantic's model_dump() converts a model instance into a plain dictionary. The ** operator unpacks that dictionary as keyword arguments, so ConnectionResponse receives all the fields at once. This avoids manually repeating every field name.
- You won't see a change yet. Next you will add the endpoint that lists all connections.
- Below the add_connection function, add the list endpoint:
@app.get("/connections", response_model=list[ConnectionResponse])
def list_connections():
db = get_db()
# Fetch all connection rows from the database
rows = db.execute("SELECT * FROM connections").fetchall()
# Convert each row tuple into a ConnectionResponse object
return [
ConnectionResponse(
id=row[0],
name=row[1],
company=row[2],
title=row[3],
email=row[4],
relationship_strength=row[5],
notes=row[6],
)
for row in rows
]What does this code do?
The response_model=list[ConnectionResponse] tells FastAPI that this endpoint returns a JSON array of connection objects. The list comprehension transforms raw database tuples into typed Pydantic models, which FastAPI then serializes to JSON automatically.
- Save main.py. The dev server auto-reloads on save.
- Refresh the Swagger docs page at http://127.0.0.1:8000/docs.
- You should see POST /connections and GET /connections listed alongside the profile endpoints.
- Test POST /connections by expanding it, clicking Try it out, entering a sample connection (name, company, title), and clicking Execute.
Not seeing the new endpoints?
Check your terminal for error messages. If you see a syntax error, make sure the add_connection and list_connections functions are outside the get_profile function (not accidentally indented inside it). Each function should start at the same indentation level as create_profile.
Help me fix my connection endpoints
Implement LinkedIn CSV parsing
Adding connections one at a time works, but most people have hundreds of contacts. LinkedIn lets you export your connections as a CSV file. This endpoint parses that export and bulk-imports your network.
- In main.py, add the CSV upload endpoint between the add_connection function and the list_connections function:
@app.post("/connections/upload-csv")
def upload_connections_csv(file: UploadFile = File(...)):
import csv
import io
# Read the uploaded file and decode from bytes to string
content = file.file.read().decode("utf-8")
reader = csv.DictReader(io.StringIO(content))
db = get_db()
count = 0
# Parse each row using LinkedIn's column headers
for row in reader:
first_name = row.get("First Name", "")
last_name = row.get("Last Name", "")
name = f"{first_name} {last_name}".strip()
company = row.get("Company", "")
title = row.get("Position", "")
email = row.get("Email Address", "")
if name:
db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(name, company, title, email, "medium", ""),
)
count += 1
db.commit()
return {"message": f"Imported {count} connections"}How does the CSV parsing work?
LinkedIn exports connections with specific column headers: First Name, Last Name, Company, Position, and Email Address. The csv.DictReader maps each row to a dictionary keyed by these headers.
The UploadFile parameter tells FastAPI to accept a file upload. The File(...) marker makes the file required. Each valid row (one with a non-empty name) gets inserted with a default relationship strength of medium.
- Save main.py.
- Refresh the Swagger docs at http://127.0.0.1:8000/docs.
- You should see POST /connections/upload-csv with a file upload button when you expand it.
Testing without a real LinkedIn export
Create a file called test_connections.csv with this content to test the upload:
First Name,Last Name,Email Address,Company,Position
Jane,Smith,jane@example.com,Stripe,Engineering Manager
Bob,Johnson,bob@example.com,Airbnb,Senior DeveloperUpload this file through the Swagger docs to verify the parser works.
Getting a 422 validation error on upload?
Make sure you're uploading a file (not pasting text). In Swagger, the upload-csv endpoint should show a Choose File button. If you see a text box instead, refresh the page. The 422 means FastAPI expected a file but received something else.
Help me fix my CSV upload endpoint
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py with all five endpoints.
ⓧ I'd like to double check the full code
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profile
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)
@app.post("/profile", response_model=ProfileResponse)
def create_profile(profile: ProfileCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO profiles (name, skills, target_titles, preferred_locations, experience_level, resume_summary)
VALUES (?, ?, ?, ?, ?, ?)""",
(
profile.name,
json.dumps(profile.skills),
json.dumps(profile.target_titles),
json.dumps(profile.preferred_locations),
profile.experience_level,
profile.resume_summary,
),
)
db.commit()
return ProfileResponse(
id=cursor.lastrowid,
name=profile.name,
skills=profile.skills,
target_titles=profile.target_titles,
preferred_locations=profile.preferred_locations,
experience_level=profile.experience_level,
resume_summary=profile.resume_summary,
)
@app.get("/profile/{profile_id}", response_model=ProfileResponse)
def get_profile(profile_id: int):
db = get_db()
row = db.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profile not found")
return ProfileResponse(
id=row[0],
name=row[1],
skills=json.loads(row[2]),
target_titles=json.loads(row[3]),
preferred_locations=json.loads(row[4]),
experience_level=row[5],
resume_summary=row[6],
)
@app.post("/connections", response_model=ConnectionResponse)
def add_connection(connection: ConnectionCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(
connection.name,
connection.company,
connection.title,
connection.email,
connection.relationship_strength,
connection.notes,
),
)
db.commit()
return ConnectionResponse(id=cursor.lastrowid, **connection.model_dump())
@app.post("/connections/upload-csv")
def upload_connections_csv(file: UploadFile = File(...)):
import csv
import io
content = file.file.read().decode("utf-8")
reader = csv.DictReader(io.StringIO(content))
db = get_db()
count = 0
for row in reader:
first_name = row.get("First Name", "")
last_name = row.get("Last Name", "")
name = f"{first_name} {last_name}".strip()
company = row.get("Company", "")
title = row.get("Position", "")
email = row.get("Email Address", "")
if name:
db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(name, company, title, email, "medium", ""),
)
count += 1
db.commit()
return {"message": f"Imported {count} connections"}
@app.get("/connections", response_model=list[ConnectionResponse])
def list_connections():
db = get_db()
rows = db.execute("SELECT * FROM connections").fetchall()
return [
ConnectionResponse(
id=row[0],
name=row[1],
company=row[2],
title=row[3],
email=row[4],
relationship_strength=row[5],
notes=row[6],
)
for row in rows
]Your API can now create profiles and manage connections. Next up, you will build the job scraper that fetches real listings from company career pages.
Build the Job Scraper
Your profile and connections are stored in the database. Now you need actual job data to match against them.
Many companies use Greenhouse to manage their career pages. Greenhouse provides a free, public JSON API that returns structured job listings. Instead of scraping HTML (which breaks when a site redesigns), you will call this API directly and get clean data back every time.
In this step, get ready to:
- Create a Greenhouse API scraper that fetches real job listings.
- Add company tracking endpoints to store which companies to monitor.
- Build a scrape endpoint with duplicate detection.
Create the Greenhouse API scraper
The Greenhouse Job Board API lives at a predictable URL for every company. If a company's board token is stripe, its jobs are at https://boards-api.greenhouse.io/v1/boards/stripe/jobs. You can find a company's board token in their careers page URL.
- Create a new file called scraper.py in your referral-radar folder and add the following code:
import httpx
def scrape_greenhouse_jobs(board_token: str, company_name: str) -> list[dict]:
# Build the Greenhouse API URL using the company's board token
url = f"https://boards-api.greenhouse.io/v1/boards/{board_token}/jobs"
# Request job listings with full descriptions included
response = httpx.get(url, params={"content": "true"}, timeout=30.0)
if response.status_code != 200:
return []
data = response.json()
jobs = []
for job in data.get("jobs", []):
# Extract the location name from the nested location object
location_name = ""
if job.get("location"):
location_name = job["location"].get("name", "")
# Get the first department name if available
departments = job.get("departments", [])
department_name = departments[0]["name"] if departments else ""
# Truncate long descriptions to keep database rows manageable
description = job.get("content", "")
if len(description) > 5000:
description = description[:5000]
jobs.append(
{
"company": company_name,
"title": job.get("title", ""),
"location": location_name,
"job_url": job.get("absolute_url", ""),
"department": department_name,
"description": description,
"date_posted": job.get("updated_at", ""),
"employment_type": "Full-time",
"source_url": f"https://boards-api.greenhouse.io/v1/boards/{board_token}/jobs",
}
)
return jobsWhat does this code do?
The scrape_greenhouse_jobs function takes a board token and company name, then calls the Greenhouse public API. The content=true parameter tells the API to include full job descriptions in the response.
The function normalizes every job into a consistent dictionary format. It extracts nested fields like location.name and departments[0].name into flat strings. Descriptions longer than 5000 characters get truncated to keep the database lean.
Why a public API instead of HTML scraping?
HTML scraping breaks whenever a company redesigns their page. The Greenhouse API returns structured JSON that stays consistent regardless of how the careers page looks. Companies like Stripe, Airbnb, Notion, and Figma all use Greenhouse.
The board token is just the slug in a company's careers URL. For example, boards.greenhouse.io/stripe means the token is stripe.
- Save scraper.py.
Seeing a red underline on httpx?
Make sure your virtual environment is active and that you installed httpx in the previous step. If the import shows an error, run pip install httpx again inside your virtual environment.
help me fix my httpx import error in scraper.py
Add company tracking endpoints
Before you can scrape a company's jobs, you need a way to register which companies to track. Each company record stores its name, careers URL, and Greenhouse board token.
- In main.py, find the list_connections function. Add the following two endpoints below it:
@app.post("/companies", response_model=CompanyResponse)
def add_company(company: CompanyCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO companies (name, careers_url, greenhouse_board_token)
VALUES (?, ?, ?)""",
(company.name, company.careers_url, company.greenhouse_board_token),
)
db.commit()
return CompanyResponse(id=cursor.lastrowid, **company.model_dump())
@app.get("/companies", response_model=list[CompanyResponse])
def list_companies():
db = get_db()
rows = db.execute("SELECT * FROM companies").fetchall()
return [
CompanyResponse(
id=row[0], name=row[1], careers_url=row[2], greenhouse_board_token=row[3]
)
for row in rows
]What does this code do?
These two endpoints let you register and list companies to monitor. POST /companies accepts a company name, careers URL, and optional Greenhouse board token. GET /companies returns all tracked companies.
The CompanyCreate and CompanyResponse models (already defined in models.py) handle validation automatically.
- Save main.py.
Build the scrape endpoint with duplicate detection
The scrape endpoint ties everything together. It looks up a tracked company, calls the scraper, and stores only new jobs. Duplicate detection prevents the same listing from piling up every time you scrape.
- In main.py, add the following endpoint below the list_companies function:
@app.post("/scrape/{company_id}")
def scrape_company(company_id: int):
db = get_db()
# Look up the company by ID
company = db.execute(
"SELECT * FROM companies WHERE id = ?", (company_id,)
).fetchone()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
board_token = company[3]
if not board_token:
raise HTTPException(
status_code=400, detail="Company has no Greenhouse board token"
)
# Fetch jobs from the Greenhouse API
jobs = scrape_greenhouse_jobs(board_token, company[1])
new_count = 0
for job in jobs:
# Check for duplicates using company + title + location
existing = db.execute(
"SELECT id FROM jobs WHERE company = ? AND title = ? AND location = ?",
(job["company"], job["title"], job["location"]),
).fetchone()
if not existing:
db.execute(
"""INSERT INTO jobs (company, title, location, job_url, department, description, date_posted, employment_type, source_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
job["company"],
job["title"],
job["location"],
job["job_url"],
job["department"],
job["description"],
job["date_posted"],
job["employment_type"],
job["source_url"],
),
)
new_count += 1
db.commit()
return {"message": f"Scraped {len(jobs)} jobs, {new_count} new"}How does duplicate detection work?
Before inserting each job, the code checks if a row already exists with the same company, title, and location combination. If a match is found, it skips the insert. This means you can scrape the same company multiple times without creating duplicate entries.
The endpoint returns both the total jobs found and how many were genuinely new. If you see "Scraped 50 jobs, 0 new" on a second run, duplicate detection is working.
- Save main.py.
- Switch back to your terminal and make sure your server is running with fastapi dev main.py.
- Open http://127.0.0.1:8000/docs in your browser. You should see the new POST /companies, GET /companies, and POST /scrape/{company_id} endpoints listed in the Swagger UI.
Server won't start?
If you see an import error for scraper, make sure your scraper.py file is saved in the same folder as main.py. The from scraper import scrape_greenhouse_jobs line in main.py expects both files to be siblings.
help me fix my FastAPI server startup error after adding the scraper
Let's test the full scraping flow. You will register a company, trigger a scrape, and confirm jobs appear.
- In the Swagger UI, expand POST /companies and click Try it out.
- Paste this JSON into the request body:
{
"name": "Stripe",
"careers_url": "https://stripe.com/jobs",
"greenhouse_board_token": "stripe"
}- Click Execute. You should see a response with "id": 1 confirming the company was saved.
- Now expand POST /scrape/{company_id} and click Try it out.
- Enter 1 for the company_id parameter and click Execute.
You should see a response like "Scraped 47 jobs, 47 new" (the exact number depends on how many open roles the company has). The scraper pulled real job listings from Greenhouse and stored them in your database.
Try scraping again
Execute the same scrape request a second time. You should see the same total count but 0 new jobs. This confirms your duplicate detection is working correctly.
✔️ Awesome, I've got everything!
Great. Double check you have saved both scraper.py and main.py.
ⓧ I'd like to double check the full code
Here is the complete scraper.py:
import httpx
def scrape_greenhouse_jobs(board_token: str, company_name: str) -> list[dict]:
url = f"https://boards-api.greenhouse.io/v1/boards/{board_token}/jobs"
response = httpx.get(url, params={"content": "true"}, timeout=30.0)
if response.status_code != 200:
return []
data = response.json()
jobs = []
for job in data.get("jobs", []):
location_name = ""
if job.get("location"):
location_name = job["location"].get("name", "")
departments = job.get("departments", [])
department_name = departments[0]["name"] if departments else ""
description = job.get("content", "")
if len(description) > 5000:
description = description[:5000]
jobs.append(
{
"company": company_name,
"title": job.get("title", ""),
"location": location_name,
"job_url": job.get("absolute_url", ""),
"department": department_name,
"description": description,
"date_posted": job.get("updated_at", ""),
"employment_type": "Full-time",
"source_url": f"https://boards-api.greenhouse.io/v1/boards/{board_token}/jobs",
}
)
return jobsHere is the complete main.py up to this step:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profile
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)
@app.post("/profile", response_model=ProfileResponse)
def create_profile(profile: ProfileCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO profiles (name, skills, target_titles, preferred_locations, experience_level, resume_summary)
VALUES (?, ?, ?, ?, ?, ?)""",
(
profile.name,
json.dumps(profile.skills),
json.dumps(profile.target_titles),
json.dumps(profile.preferred_locations),
profile.experience_level,
profile.resume_summary,
),
)
db.commit()
return ProfileResponse(
id=cursor.lastrowid,
name=profile.name,
skills=profile.skills,
target_titles=profile.target_titles,
preferred_locations=profile.preferred_locations,
experience_level=profile.experience_level,
resume_summary=profile.resume_summary,
)
@app.get("/profile/{profile_id}", response_model=ProfileResponse)
def get_profile(profile_id: int):
db = get_db()
row = db.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profile not found")
return ProfileResponse(
id=row[0],
name=row[1],
skills=json.loads(row[2]),
target_titles=json.loads(row[3]),
preferred_locations=json.loads(row[4]),
experience_level=row[5],
resume_summary=row[6],
)
@app.post("/connections", response_model=ConnectionResponse)
def add_connection(connection: ConnectionCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(
connection.name,
connection.company,
connection.title,
connection.email,
connection.relationship_strength,
connection.notes,
),
)
db.commit()
return ConnectionResponse(id=cursor.lastrowid, **connection.model_dump())
@app.post("/connections/upload-csv")
def upload_connections_csv(file: UploadFile = File(...)):
import csv
import io
content = file.file.read().decode("utf-8")
reader = csv.DictReader(io.StringIO(content))
db = get_db()
count = 0
for row in reader:
first_name = row.get("First Name", "")
last_name = row.get("Last Name", "")
name = f"{first_name} {last_name}".strip()
company = row.get("Company", "")
title = row.get("Position", "")
email = row.get("Email Address", "")
if name:
db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(name, company, title, email, "medium", ""),
)
count += 1
db.commit()
return {"message": f"Imported {count} connections"}
@app.get("/connections", response_model=list[ConnectionResponse])
def list_connections():
db = get_db()
rows = db.execute("SELECT * FROM connections").fetchall()
return [
ConnectionResponse(
id=row[0],
name=row[1],
company=row[2],
title=row[3],
email=row[4],
relationship_strength=row[5],
notes=row[6],
)
for row in rows
]
@app.post("/companies", response_model=CompanyResponse)
def add_company(company: CompanyCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO companies (name, careers_url, greenhouse_board_token)
VALUES (?, ?, ?)""",
(company.name, company.careers_url, company.greenhouse_board_token),
)
db.commit()
return CompanyResponse(id=cursor.lastrowid, **company.model_dump())
@app.get("/companies", response_model=list[CompanyResponse])
def list_companies():
db = get_db()
rows = db.execute("SELECT * FROM companies").fetchall()
return [
CompanyResponse(
id=row[0], name=row[1], careers_url=row[2], greenhouse_board_token=row[3]
)
for row in rows
]
@app.post("/scrape/{company_id}")
def scrape_company(company_id: int):
db = get_db()
company = db.execute(
"SELECT * FROM companies WHERE id = ?", (company_id,)
).fetchone()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
board_token = company[3]
if not board_token:
raise HTTPException(
status_code=400, detail="Company has no Greenhouse board token"
)
jobs = scrape_greenhouse_jobs(board_token, company[1])
new_count = 0
for job in jobs:
existing = db.execute(
"SELECT id FROM jobs WHERE company = ? AND title = ? AND location = ?",
(job["company"], job["title"], job["location"]),
).fetchone()
if not existing:
db.execute(
"""INSERT INTO jobs (company, title, location, job_url, department, description, date_posted, employment_type, source_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
job["company"],
job["title"],
job["location"],
job["job_url"],
job["department"],
job["description"],
job["date_posted"],
job["employment_type"],
job["source_url"],
),
)
new_count += 1
db.commit()
return {"message": f"Scraped {len(jobs)} jobs, {new_count} new"}Your scraper is pulling real job data from live career pages. Next up, you will connect Claude AI to score how well each job matches your profile.
Build the AI Job Matching Engine
You now have a scraper that pulls real jobs from Greenhouse career pages into your database. The next question is: which of those jobs actually match your skills?
In this step, you will build an AI matching engine that sends each scraped job description to Claude Haiku 4.5 alongside a user's profile. Claude scores how well the candidate fits each role from 0 to 100 and explains its reasoning in one sentence.
In this step, get ready to:
- Set up your Anthropic API key for Claude access.
- Build a matching engine that scores jobs against a profile.
- Create API endpoints to trigger matching and retrieve results.
Set up your Anthropic API key
The anthropic Python SDK reads your API key from an environment variable called ANTHROPIC_API_KEY. You need to set this before the matching engine can call Claude.
- If you do not have an Anthropic API key yet, create one at console.anthropic.com.
- Set the environment variable in the same terminal where you run your server by running this command:
macOS
export ANTHROPIC_API_KEY=your-api-key-hereWindows
$env:ANTHROPIC_API_KEY="your-api-key-here"- Replace your-api-key-here with the key you copied from the Anthropic console.
Why an environment variable?
Storing secrets in environment variables keeps them out of your source code. The anthropic SDK automatically reads ANTHROPIC_API_KEY when you create a client, so you never hard-code the key in your Python files.
Build the matching engine
The matching engine takes a user profile and a list of job rows from the database. For each job, it constructs a prompt asking Claude to compare the candidate against the job posting and return a score.
- Open matcher.py in Cursor and add the following code:
import anthropic
import json
def match_jobs_for_profile(profile: dict, job_rows: list) -> list[dict]:
# Create the Anthropic client (reads ANTHROPIC_API_KEY automatically)
client = anthropic.Anthropic()
results = []
# Build a text summary of the candidate's profile
profile_summary = f"""
Name: {profile['name']}
Skills: {', '.join(profile['skills'])}
Target Titles: {', '.join(profile['target_titles'])}
Preferred Locations: {', '.join(profile['preferred_locations'])}
Experience Level: {profile['experience_level']}
Resume Summary: {profile['resume_summary']}
"""
for row in job_rows:
job_id = row[0]
job_title = row[2]
job_location = row[3]
job_description = row[6] or ""
# Truncate long descriptions to keep token costs low
if len(job_description) > 3000:
job_description = job_description[:3000]
job_summary = f"Title: {job_title}\nLocation: {job_location}\nDescription: {job_description}"
# Prompt asks Claude to return a JSON score and reasoning
prompt = f"""You are a job matching assistant. Compare this candidate profile with the job posting and return a JSON object with two fields:
- "score": an integer from 0 to 100 representing how well the candidate matches the job
- "reasoning": a brief one-sentence explanation
Candidate Profile:
{profile_summary}
Job Posting:
{job_summary}
Return ONLY valid JSON, no other text."""
try:
message = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{"role": "user", "content": prompt}],
)
response_text = message.content[0].text
parsed = json.loads(response_text)
results.append(
{
"job_id": job_id,
"score": int(parsed.get("score", 0)),
"reasoning": parsed.get("reasoning", ""),
}
)
except (json.JSONDecodeError, IndexError, KeyError):
# If Claude's response isn't valid JSON, record a zero score
results.append(
{
"job_id": job_id,
"score": 0,
"reasoning": "Failed to parse AI response",
}
)
return resultsWhat does this code do?
The function loops through every job row from the database and sends a structured prompt to Claude Haiku 4.5. The prompt includes the candidate's skills, target titles, locations, and resume summary alongside the job title, location, and description.
Claude returns a JSON object with a score (0-100) and reasoning (one sentence). The try/except block handles cases where Claude's response is not valid JSON by defaulting to a score of 0.
Why Claude Haiku 4.5 instead of a larger model?
Job matching is a straightforward text comparison task. Haiku 4.5 costs $1 per million input tokens and $5 per million output tokens. That is roughly 10x cheaper than larger models. For scoring and one-sentence reasoning, Haiku is fast and more than capable.
- Save matcher.py.
Seeing an import error for anthropic?
Make sure you installed the anthropic package in your virtual environment. Run pip install anthropic again if needed. Also confirm your virtual environment is active (you should see (venv) in your terminal prompt).
help me fix an anthropic import error in my Python project
✔️ Awesome, I've got everything!
Great. Make sure you have saved matcher.py before moving on.
ⓧ I'd like to double check the full code
import anthropic
import json
def match_jobs_for_profile(profile: dict, job_rows: list) -> list[dict]:
client = anthropic.Anthropic()
results = []
profile_summary = f"""
Name: {profile['name']}
Skills: {', '.join(profile['skills'])}
Target Titles: {', '.join(profile['target_titles'])}
Preferred Locations: {', '.join(profile['preferred_locations'])}
Experience Level: {profile['experience_level']}
Resume Summary: {profile['resume_summary']}
"""
for row in job_rows:
job_id = row[0]
job_title = row[2]
job_location = row[3]
job_description = row[6] or ""
if len(job_description) > 3000:
job_description = job_description[:3000]
job_summary = f"Title: {job_title}\nLocation: {job_location}\nDescription: {job_description}"
prompt = f"""You are a job matching assistant. Compare this candidate profile with the job posting and return a JSON object with two fields:
- "score": an integer from 0 to 100 representing how well the candidate matches the job
- "reasoning": a brief one-sentence explanation
Candidate Profile:
{profile_summary}
Job Posting:
{job_summary}
Return ONLY valid JSON, no other text."""
try:
message = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{"role": "user", "content": prompt}],
)
response_text = message.content[0].text
parsed = json.loads(response_text)
results.append(
{
"job_id": job_id,
"score": int(parsed.get("score", 0)),
"reasoning": parsed.get("reasoning", ""),
}
)
except (json.JSONDecodeError, IndexError, KeyError):
results.append(
{
"job_id": job_id,
"score": 0,
"reasoning": "Failed to parse AI response",
}
)
return resultsCreate the matching endpoints
With the matching engine ready, you need two endpoints. One triggers the matching process and stores results. The other retrieves stored match scores so you can see which jobs fit best.
- In main.py, scroll to the bottom of the file (below the scrape_company function) and add the run matching endpoint:
@app.post("/match/{profile_id}")
def run_matching(profile_id: int):
db = get_db()
# Fetch the profile from the database
profile_row = db.execute(
"SELECT * FROM profiles WHERE id = ?", (profile_id,)
).fetchone()
if not profile_row:
raise HTTPException(status_code=404, detail="Profile not found")
# Reconstruct the profile as a dictionary for the matcher
profile_data = {
"name": profile_row[1],
"skills": json.loads(profile_row[2]),
"target_titles": json.loads(profile_row[3]),
"preferred_locations": json.loads(profile_row[4]),
"experience_level": profile_row[5],
"resume_summary": profile_row[6],
}
# Get all scraped jobs and run them through the matcher
jobs = db.execute("SELECT * FROM jobs").fetchall()
results = match_jobs_for_profile(profile_data, jobs)
# Upsert each result into the matches table
for result in results:
db.execute(
"""INSERT OR REPLACE INTO matches (profile_id, job_id, score, reasoning)
VALUES (?, ?, ?, ?)""",
(profile_id, result["job_id"], result["score"], result["reasoning"]),
)
db.commit()
return {"message": f"Matched {len(results)} jobs"}What does this endpoint do?
When you call POST /match/{profile_id}, the endpoint fetches the profile from the database, grabs all scraped jobs, and passes both to the matching engine. Claude scores each job and the results are saved using INSERT OR REPLACE so re-running the matcher updates existing scores instead of creating duplicates.
- Below the run_matching function, add the endpoint to retrieve match results:
@app.get("/matches/{profile_id}", response_model=list[MatchResponse])
def get_matches(profile_id: int, min_score: int = 0):
db = get_db()
# Join matches with jobs to get full job details in one query
rows = db.execute(
"""SELECT m.job_id, m.score, m.reasoning, j.company, j.title, j.location, j.job_url
FROM matches m JOIN jobs j ON m.job_id = j.id
WHERE m.profile_id = ? AND m.score >= ?
ORDER BY m.score DESC""",
(profile_id, min_score),
).fetchall()
return [
MatchResponse(
job_id=row[0],
score=row[1],
reasoning=row[2],
company=row[3],
title=row[4],
location=row[5],
job_url=row[6],
)
for row in rows
]What does this endpoint do?
The GET /matches/{profile_id} endpoint retrieves all stored match scores for a profile. The optional min_score query parameter filters out low-scoring jobs. Results come back sorted highest score first, so the best matches appear at the top.
- Save main.py.
- Restart your dev server by running:
fastapi dev- Open http://127.0.0.1:8000/docs in your browser. You should see the new POST /match/{profile_id} and GET /matches/{profile_id} endpoints listed in the Swagger UI.
Server won't start?
Check that the match_jobs_for_profile import at the top of main.py resolves correctly. Make sure matcher.py is in the same directory as main.py and that you saved both files.
help me debug why my FastAPI server won't start after adding the matching endpoints
✔️ Awesome, I've got everything!
Great. Make sure both matcher.py and main.py are saved before continuing.
ⓧ I'd like to double check the full code
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profile
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)
@app.post("/profile", response_model=ProfileResponse)
def create_profile(profile: ProfileCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO profiles (name, skills, target_titles, preferred_locations, experience_level, resume_summary)
VALUES (?, ?, ?, ?, ?, ?)""",
(
profile.name,
json.dumps(profile.skills),
json.dumps(profile.target_titles),
json.dumps(profile.preferred_locations),
profile.experience_level,
profile.resume_summary,
),
)
db.commit()
return ProfileResponse(
id=cursor.lastrowid,
name=profile.name,
skills=profile.skills,
target_titles=profile.target_titles,
preferred_locations=profile.preferred_locations,
experience_level=profile.experience_level,
resume_summary=profile.resume_summary,
)
@app.get("/profile/{profile_id}", response_model=ProfileResponse)
def get_profile(profile_id: int):
db = get_db()
row = db.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profile not found")
return ProfileResponse(
id=row[0],
name=row[1],
skills=json.loads(row[2]),
target_titles=json.loads(row[3]),
preferred_locations=json.loads(row[4]),
experience_level=row[5],
resume_summary=row[6],
)
@app.post("/connections", response_model=ConnectionResponse)
def add_connection(connection: ConnectionCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(
connection.name,
connection.company,
connection.title,
connection.email,
connection.relationship_strength,
connection.notes,
),
)
db.commit()
return ConnectionResponse(id=cursor.lastrowid, **connection.model_dump())
@app.post("/connections/upload-csv")
def upload_connections_csv(file: UploadFile = File(...)):
import csv
import io
content = file.file.read().decode("utf-8")
reader = csv.DictReader(io.StringIO(content))
db = get_db()
count = 0
for row in reader:
first_name = row.get("First Name", "")
last_name = row.get("Last Name", "")
name = f"{first_name} {last_name}".strip()
company = row.get("Company", "")
title = row.get("Position", "")
email = row.get("Email Address", "")
if name:
db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(name, company, title, email, "medium", ""),
)
count += 1
db.commit()
return {"message": f"Imported {count} connections"}
@app.get("/connections", response_model=list[ConnectionResponse])
def list_connections():
db = get_db()
rows = db.execute("SELECT * FROM connections").fetchall()
return [
ConnectionResponse(
id=row[0],
name=row[1],
company=row[2],
title=row[3],
email=row[4],
relationship_strength=row[5],
notes=row[6],
)
for row in rows
]
@app.post("/companies", response_model=CompanyResponse)
def add_company(company: CompanyCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO companies (name, careers_url, greenhouse_board_token)
VALUES (?, ?, ?)""",
(company.name, company.careers_url, company.greenhouse_board_token),
)
db.commit()
return CompanyResponse(id=cursor.lastrowid, **company.model_dump())
@app.get("/companies", response_model=list[CompanyResponse])
def list_companies():
db = get_db()
rows = db.execute("SELECT * FROM companies").fetchall()
return [
CompanyResponse(
id=row[0], name=row[1], careers_url=row[2], greenhouse_board_token=row[3]
)
for row in rows
]
@app.post("/scrape/{company_id}")
def scrape_company(company_id: int):
db = get_db()
company = db.execute(
"SELECT * FROM companies WHERE id = ?", (company_id,)
).fetchone()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
board_token = company[3]
if not board_token:
raise HTTPException(
status_code=400, detail="Company has no Greenhouse board token"
)
jobs = scrape_greenhouse_jobs(board_token, company[1])
new_count = 0
for job in jobs:
existing = db.execute(
"SELECT id FROM jobs WHERE company = ? AND title = ? AND location = ?",
(job["company"], job["title"], job["location"]),
).fetchone()
if not existing:
db.execute(
"""INSERT INTO jobs (company, title, location, job_url, department, description, date_posted, employment_type, source_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
job["company"],
job["title"],
job["location"],
job["job_url"],
job["department"],
job["description"],
job["date_posted"],
job["employment_type"],
job["source_url"],
),
)
new_count += 1
db.commit()
return {"message": f"Scraped {len(jobs)} jobs, {new_count} new"}
@app.post("/match/{profile_id}")
def run_matching(profile_id: int):
db = get_db()
profile_row = db.execute(
"SELECT * FROM profiles WHERE id = ?", (profile_id,)
).fetchone()
if not profile_row:
raise HTTPException(status_code=404, detail="Profile not found")
profile_data = {
"name": profile_row[1],
"skills": json.loads(profile_row[2]),
"target_titles": json.loads(profile_row[3]),
"preferred_locations": json.loads(profile_row[4]),
"experience_level": profile_row[5],
"resume_summary": profile_row[6],
}
jobs = db.execute("SELECT * FROM jobs").fetchall()
results = match_jobs_for_profile(profile_data, jobs)
for result in results:
db.execute(
"""INSERT OR REPLACE INTO matches (profile_id, job_id, score, reasoning)
VALUES (?, ?, ?, ?)""",
(profile_id, result["job_id"], result["score"], result["reasoning"]),
)
db.commit()
return {"message": f"Matched {len(results)} jobs"}
@app.get("/matches/{profile_id}", response_model=list[MatchResponse])
def get_matches(profile_id: int, min_score: int = 0):
db = get_db()
rows = db.execute(
"""SELECT m.job_id, m.score, m.reasoning, j.company, j.title, j.location, j.job_url
FROM matches m JOIN jobs j ON m.job_id = j.id
WHERE m.profile_id = ? AND m.score >= ?
ORDER BY m.score DESC""",
(profile_id, min_score),
).fetchall()
return [
MatchResponse(
job_id=row[0],
score=row[1],
reasoning=row[2],
company=row[3],
title=row[4],
location=row[5],
job_url=row[6],
)
for row in rows
]Your matching engine is live. Next up, you will connect your personal network to these match results so you can see which high-scoring jobs have a referral path through your connections.
Build Referral Mapping and Company Tracking
In the last step, you built an AI matching engine that scores every job against your profile. That's powerful on its own, but the real value of ReferralRadar is knowing which high-scoring jobs you can get referred to.
In this step, you'll connect the final dots. You'll cross-reference your connections with companies that have matching jobs, surfacing only the opportunities where you have an "in."
In this step, get ready to:
- Test the company tracking endpoints via Swagger.
- Build a referral mapping endpoint that joins matches, jobs, and connections.
- Verify the full referral pipeline end-to-end.
Test the company tracking endpoints
Your API already has endpoints for adding and listing companies to monitor. Before building the referral logic, you'll add a test company through Swagger to have data ready for the referral query.
- Switch back to your browser tab with Swagger docs at http://127.0.0.1:8000/docs.
- Expand the POST /companies endpoint.
- Click Try it out.
- Replace the request body with the following JSON:
{
"name": "Stripe",
"careers_url": "https://stripe.com/jobs",
"greenhouse_board_token": "stripe"
}- Click Execute.
- You should see a 200 response with the company ID in the response body.
What is a board token?
The greenhouse_board_token is the slug visible in a company's Greenhouse careers page URL. For example, boards.greenhouse.io/stripe means the token is stripe.
You can find these tokens by visiting any company's careers page that uses Greenhouse and looking at the URL.
Build the referral-ready jobs endpoint
The referral mapping logic is a SQL query that joins three tables: matches (your scored jobs), jobs (the job details), and connections (your network). If a connection works at the same company as a matched job, that job is "referral-ready."
- In main.py, scroll to the bottom of the file after the get_matches function.
- Add the referral mapping endpoint by pasting this code at the end of the file:
@app.get("/referrals/{profile_id}", response_model=list[ReferralJobResponse])
def get_referral_jobs(profile_id: int, min_score: int = 60):
db = get_db()
rows = db.execute(
"""SELECT m.job_id, m.score, m.reasoning, j.company, j.title, j.location, j.job_url,
c.name as connection_name, c.title as connection_title, c.email as connection_email
FROM matches m
JOIN jobs j ON m.job_id = j.id
JOIN connections c ON LOWER(c.company) = LOWER(j.company)
WHERE m.profile_id = ? AND m.score >= ?
ORDER BY m.score DESC""",
(profile_id, min_score),
).fetchall()
return [
ReferralJobResponse(
job_id=row[0],
score=row[1],
reasoning=row[2],
company=row[3],
title=row[4],
location=row[5],
job_url=row[6],
connection_name=row[7],
connection_title=row[8],
connection_email=row[9],
)
for row in rows
]What does this code do?
- The three-way JOIN connects matches to jobs (for job details) and connections (for your network). Only jobs where a connection's company matches the job's company appear in results.
- LOWER(c.company) = LOWER(j.company) performs a case-insensitive match so "Stripe" in your connections matches "stripe" from scraped jobs.
- min_score defaults to 60, filtering out low-relevance matches so you only see jobs worth pursuing.
- The response includes connection_name, connection_title, and connection_email so you know exactly who to reach out to.
- Save main.py.
- Your FastAPI dev server auto-reloads. Refresh Swagger docs at http://127.0.0.1:8000/docs and you should see a new GET /referrals/{profile_id} endpoint listed.
Don't see the new endpoint?
- Check your terminal for import errors. If ReferralJobResponse is missing from your imports at the top of main.py, the server will fail to start.
- Make sure the new function is at the module level (not indented inside another function).
- Confirm you saved the file. The dev server only reloads on save.
Help me debug why my /referrals endpoint isn't showing up in Swagger.
Why case-insensitive matching?
When you add connections manually or import them from a CSV, company names might be stored as "Stripe", "STRIPE", or "stripe". Scraped job data may use a different casing too.
Using LOWER() on both sides of the comparison ensures you never miss a referral opportunity because of capitalization differences.
Test the full referral pipeline
To see the referral endpoint produce results, you need data in the pipeline: a profile, connections at a tracked company, scraped jobs from that company, and match scores. Let's walk through a quick test.
- In Swagger, use POST /connections to add a test connection at a tracked company:
{
"name": "Jane Smith",
"company": "Stripe",
"title": "Engineering Manager",
"email": "jane@example.com",
"relationship_strength": "strong",
"notes": "Met at a conference"
}- Use POST /scrape/{company_id} with the company ID from earlier (likely 1) to fetch jobs from Stripe's Greenhouse page.
- Use POST /match/{profile_id} with your profile ID (likely 1) to score those jobs against your profile.
- Now test GET /referrals/{profile_id} with the same profile ID.
- You should see a list of jobs at Stripe that scored 60 or higher, each including Jane Smith's name, title, and email as the referral contact.
Adjust the minimum score
If you get an empty list, try adding ?min_score=0 to see all matched jobs regardless of score. This helps you confirm the JOIN is working even if Claude scored conservatively.
Getting an empty response?
- The connection's company field must match the company name used when scraping. If you added the company as "Stripe" but your connection says "stripe inc", the JOIN won't match.
- Make sure you ran both the scrape AND the match endpoints before testing referrals. Referrals depend on match scores existing in the database.
- Confirm your profile ID is correct by checking GET /profile/{profile_id} first.
Help me figure out why my referrals endpoint returns an empty list.
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and your server is running without errors.
ⓧ I'd like to double check the full code
Here is the complete main.py with all endpoints including the new referral mapping:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sqlite3
import json
from database import init_db, get_db
from models import (
ProfileCreate,
ProfileResponse,
ConnectionCreate,
ConnectionResponse,
CompanyCreate,
CompanyResponse,
JobResponse,
MatchResponse,
ReferralJobResponse,
)
from scraper import scrape_greenhouse_jobs
from matcher import match_jobs_for_profile
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
app = FastAPI(
title="ReferralRadar",
description="AI-powered job referral platform that scrapes career pages and matches jobs to your profile",
version="1.0.0",
lifespan=lifespan,
)
@app.post("/profile", response_model=ProfileResponse)
def create_profile(profile: ProfileCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO profiles (name, skills, target_titles, preferred_locations, experience_level, resume_summary)
VALUES (?, ?, ?, ?, ?, ?)""",
(
profile.name,
json.dumps(profile.skills),
json.dumps(profile.target_titles),
json.dumps(profile.preferred_locations),
profile.experience_level,
profile.resume_summary,
),
)
db.commit()
return ProfileResponse(
id=cursor.lastrowid,
name=profile.name,
skills=profile.skills,
target_titles=profile.target_titles,
preferred_locations=profile.preferred_locations,
experience_level=profile.experience_level,
resume_summary=profile.resume_summary,
)
@app.get("/profile/{profile_id}", response_model=ProfileResponse)
def get_profile(profile_id: int):
db = get_db()
row = db.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profile not found")
return ProfileResponse(
id=row[0],
name=row[1],
skills=json.loads(row[2]),
target_titles=json.loads(row[3]),
preferred_locations=json.loads(row[4]),
experience_level=row[5],
resume_summary=row[6],
)
@app.post("/connections", response_model=ConnectionResponse)
def add_connection(connection: ConnectionCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(
connection.name,
connection.company,
connection.title,
connection.email,
connection.relationship_strength,
connection.notes,
),
)
db.commit()
return ConnectionResponse(id=cursor.lastrowid, **connection.model_dump())
@app.post("/connections/upload-csv")
def upload_connections_csv(file: UploadFile = File(...)):
import csv
import io
content = file.file.read().decode("utf-8")
reader = csv.DictReader(io.StringIO(content))
db = get_db()
count = 0
for row in reader:
first_name = row.get("First Name", "")
last_name = row.get("Last Name", "")
name = f"{first_name} {last_name}".strip()
company = row.get("Company", "")
title = row.get("Position", "")
email = row.get("Email Address", "")
if name:
db.execute(
"""INSERT INTO connections (name, company, title, email, relationship_strength, notes)
VALUES (?, ?, ?, ?, ?, ?)""",
(name, company, title, email, "medium", ""),
)
count += 1
db.commit()
return {"message": f"Imported {count} connections"}
@app.get("/connections", response_model=list[ConnectionResponse])
def list_connections():
db = get_db()
rows = db.execute("SELECT * FROM connections").fetchall()
return [
ConnectionResponse(
id=row[0],
name=row[1],
company=row[2],
title=row[3],
email=row[4],
relationship_strength=row[5],
notes=row[6],
)
for row in rows
]
@app.post("/companies", response_model=CompanyResponse)
def add_company(company: CompanyCreate):
db = get_db()
cursor = db.execute(
"""INSERT INTO companies (name, careers_url, greenhouse_board_token)
VALUES (?, ?, ?)""",
(company.name, company.careers_url, company.greenhouse_board_token),
)
db.commit()
return CompanyResponse(id=cursor.lastrowid, **company.model_dump())
@app.get("/companies", response_model=list[CompanyResponse])
def list_companies():
db = get_db()
rows = db.execute("SELECT * FROM companies").fetchall()
return [
CompanyResponse(
id=row[0], name=row[1], careers_url=row[2], greenhouse_board_token=row[3]
)
for row in rows
]
@app.post("/scrape/{company_id}")
def scrape_company(company_id: int):
db = get_db()
company = db.execute(
"SELECT * FROM companies WHERE id = ?", (company_id,)
).fetchone()
if not company:
raise HTTPException(status_code=404, detail="Company not found")
board_token = company[3]
if not board_token:
raise HTTPException(
status_code=400, detail="Company has no Greenhouse board token"
)
jobs = scrape_greenhouse_jobs(board_token, company[1])
new_count = 0
for job in jobs:
existing = db.execute(
"SELECT id FROM jobs WHERE company = ? AND title = ? AND location = ?",
(job["company"], job["title"], job["location"]),
).fetchone()
if not existing:
db.execute(
"""INSERT INTO jobs (company, title, location, job_url, department, description, date_posted, employment_type, source_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
job["company"],
job["title"],
job["location"],
job["job_url"],
job["department"],
job["description"],
job["date_posted"],
job["employment_type"],
job["source_url"],
),
)
new_count += 1
db.commit()
return {"message": f"Scraped {len(jobs)} jobs, {new_count} new"}
@app.post("/match/{profile_id}")
def run_matching(profile_id: int):
db = get_db()
profile_row = db.execute(
"SELECT * FROM profiles WHERE id = ?", (profile_id,)
).fetchone()
if not profile_row:
raise HTTPException(status_code=404, detail="Profile not found")
profile_data = {
"name": profile_row[1],
"skills": json.loads(profile_row[2]),
"target_titles": json.loads(profile_row[3]),
"preferred_locations": json.loads(profile_row[4]),
"experience_level": profile_row[5],
"resume_summary": profile_row[6],
}
jobs = db.execute("SELECT * FROM jobs").fetchall()
results = match_jobs_for_profile(profile_data, jobs)
for result in results:
db.execute(
"""INSERT OR REPLACE INTO matches (profile_id, job_id, score, reasoning)
VALUES (?, ?, ?, ?)""",
(profile_id, result["job_id"], result["score"], result["reasoning"]),
)
db.commit()
return {"message": f"Matched {len(results)} jobs"}
@app.get("/matches/{profile_id}", response_model=list[MatchResponse])
def get_matches(profile_id: int, min_score: int = 0):
db = get_db()
rows = db.execute(
"""SELECT m.job_id, m.score, m.reasoning, j.company, j.title, j.location, j.job_url
FROM matches m JOIN jobs j ON m.job_id = j.id
WHERE m.profile_id = ? AND m.score >= ?
ORDER BY m.score DESC""",
(profile_id, min_score),
).fetchall()
return [
MatchResponse(
job_id=row[0],
score=row[1],
reasoning=row[2],
company=row[3],
title=row[4],
location=row[5],
job_url=row[6],
)
for row in rows
]
@app.get("/referrals/{profile_id}", response_model=list[ReferralJobResponse])
def get_referral_jobs(profile_id: int, min_score: int = 60):
db = get_db()
rows = db.execute(
"""SELECT m.job_id, m.score, m.reasoning, j.company, j.title, j.location, j.job_url,
c.name as connection_name, c.title as connection_title, c.email as connection_email
FROM matches m
JOIN jobs j ON m.job_id = j.id
JOIN connections c ON LOWER(c.company) = LOWER(j.company)
WHERE m.profile_id = ? AND m.score >= ?
ORDER BY m.score DESC""",
(profile_id, min_score),
).fetchall()
return [
ReferralJobResponse(
job_id=row[0],
score=row[1],
reasoning=row[2],
company=row[3],
title=row[4],
location=row[5],
job_url=row[6],
connection_name=row[7],
connection_title=row[8],
connection_email=row[9],
)
for row in rows
]Your ReferralRadar API is now complete. You have a working pipeline that scrapes real jobs, scores them with AI, and maps your network to referral opportunities. Next up, you'll explore extending this with a live HTML dashboard in the Secret Mission.
Secret mission
Build a Live HTML Dashboard
Your API returns JSON data, but wouldn't it be nice to see your top matches at a glance? In this secret mission, you'll build a live HTML dashboard served directly by FastAPI that shows color-coded match scores and highlights which jobs have referral-ready connections.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing cloud costs. The only charges incurred were per-request Claude API usage during the matching step.
Resources you used:
- referralradar project folder on your Desktop (contains main.py, database.py, models.py, scraper.py, matcher.py, virtual environment, and SQLite database).
- ANTHROPIC_API_KEY environment variable set in your terminal session.
- Anthropic API key stored in your Anthropic console account.
Keep everything running
No action needed. Choose this if you plan to keep building on ReferralRadar or want to add more companies and connections later.
- Your project files stay on your Desktop in the referralradar folder.
- The FastAPI server only runs when you start it manually with fastapi dev. It does not run in the background.
- Your ANTHROPIC_API_KEY remains available for future sessions if you added it to your shell profile.
Pause - I'll come back to this later
Shut down the running dev server to free up your terminal. Your files and database stay intact for when you return.
- Press Ctrl+C in the terminal where your FastAPI server is running to stop it.
- Your project files, database, and virtual environment remain in the referralradar folder on your Desktop.
- To restart later, activate your virtual environment and run fastapi dev again from the project folder.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes your code, database, and scraped job data permanently.
- Press Ctrl+C in the terminal where your FastAPI server is running to stop it.
- Delete the project folder by running this command (macOS/Linux):
rm -rf ~/Desktop/referralradar- On Windows, run this instead:
Remove-Item -Recurse -Force ~\Desktop\referralradar- Remove the ANTHROPIC_API_KEY environment variable from your current session:
unset ANTHROPIC_API_KEY- If you added ANTHROPIC_API_KEY to your shell profile (.zshrc or .bashrc), open the file and delete the export ANTHROPIC_API_KEY=... line.
- Delete your API key from the Anthropic console. Go to console.anthropic.com/settings/keys, find the key you created, and click the delete icon to revoke it.
That's a Wrap!
That's a Wrap!
Nice work! You just built ReferralRadar, a full-stack AI-powered job referral platform that scrapes real career pages, matches jobs to your profile, and maps your network to referral opportunities.
You've learned how to:
- Built a complete FastAPI backend with auto-generated Swagger docs, a Greenhouse API scraper, and SQLite storage with duplicate detection.
- Integrated Claude Haiku 4.5 to intelligently score jobs 0-100 against your profile and return reasoning for each match.
- Created a referral mapping system that cross-references your personal connections with companies that have high-scoring jobs, plus a CSV importer for LinkedIn exports.
- Built a live HTML dashboard with color-coded match scores and referral-ready badges showing which connections can refer you.
Ready to quiz yourself?