Agentic Job Intelligence Pipeline (MCP + LLM)

An agent-driven pipeline that uses the Model Context Protocol to orchestrate external tools for structured data retrieval, with an LLM scoring layer that ranks unstructured job descriptions against a candidate profile. Context-window pressure is handled with a staged metadata-first retrieval strategy (pipeline.py), and the same tools are also exposed to a genuine tool-calling agent with its own planning loop (agent.py) and to a REST + WebSocket API (api.py). See docs/ for the full write-up.

Run it

pip install -r requirements.txt

python pipeline.py --benchmark     # token comparison, zero API calls
python pipeline.py --dry-run       # real MCP subprocess handshake, no LLM
export OPENAI_API_KEY=sk-...
python pipeline.py --top 8         # fixed 3-stage pipeline
python agent.py --dry-run          # agent tool discovery, no LLM calls
export OPENAI_API_KEY=sk-...
python agent.py --top 8            # tool-calling agent with a planning loop
python eval.py --prefilter-only    # stage-1 recall, deterministic half, no key needed
pytest -q                          # in-process MCP server, no key needed

uvicorn api:app --reload           # REST + WebSocket layer, http://localhost:8000
curl localhost:8000/health
curl -X POST localhost:8000/rank -H 'content-type: application/json' -d '{"use_llm": false}'

Measured result

150-job corpus, shortlist of 8. prefilter() applies tag/title, seniority (drop senior when candidate years < 4), and location (candidate's preferred city, alias-normalised, or remote) gates, which cut the survivor count from 150 to 31:

Strategy	Prompt tokens	vs naive
A — send all 150 full descriptions	67,360	—
B — metadata-first, then fetch 8	13,802	4.9× cheaper
C — prefilter → metadata → fetch 8	6,258	10.8× cheaper

Reproduce with python pipeline.py --benchmark. Real numbers from this repo's data/jobs.json today, not placeholders. Token counts use tiktoken's o200k_base encoder (what gpt-4o / gpt-4o-mini actually use) — exact, not estimated. The old chars ÷ 4 heuristic overestimated the naive-strategy cost by 17.4% on this corpus; benchmark()'s heuristic_vs_real_tokens field reproduces that comparison.

Architecture

   MCP SERVER (stdio subprocess)              MCP CLIENT / pipeline.py
   ---------------------------------          ------------------------------------
   tool  list_jobs        -> metadata  <----  Stage 0  prefilter()   [0 tokens]
   tool  get_job_details  -> full text        Stage 1  shortlist     [~5k tokens]
   tool  get_candidate_profile                Stage 2  score         [~4k tokens]
   tool  corpus_stats
   resource  jobs://schema                    Meter tracks tokens per stage
   prompt    rank_jobs

The staged retrieval argument

Naive: hand every full description to the model and ask it to rank. Three problems.

1. Cost — 79k prompt tokens per run, and it grows linearly with the corpus.
2. Ceiling — past a few hundred postings it exceeds the context window outright. Not slow: impossible.
3. Quality — long-context recall degrades in the middle of a large prompt, so the ranking gets worse as you add more candidates.

Staged retrieval, cheapest filter first:

Stage	Mechanism	Cost	Why here
0	Deterministic tag/title/location filter in Python	free	Never let a model read what if could discard. 150 → 91.
1	LLM sees ~55 tokens of metadata per job, picks top 8	~5k	High-recall screen. Instructed to over-include, because stage 2 can reject.
2	Full descriptions for the 8 survivors only	~4k	Full fidelity, paid for once, only where it changes the answer.

The generalisable principle — and the thing to say out loud in an interview — is cascade by cost: order your filters cheapest-first, and set each stage's threshold for recall rather than precision, because a later stage can still reject but nothing can recover what an early stage dropped.

Agentic layer (agent.py)

pipeline.py is a fixed script: prefilter, then always shortlist, then always score. agent.py hands the model the same MCP tools via OpenAI function calling and lets it plan its own path — a genuine tool-calling agent, not a hardcoded sequence:

• Structured final answer as a tool call. The agent doesn't "answer in JSON and hope" — finishing means calling a synthetic submit_rankings tool whose parameter schema is schemas.RankingResult. Invalid arguments come back as a validation error the model can read and correct, for a bounded number of retries.
• Tool failures degrade, they don't crash. Any MCP tool exception becomes a normal {"error": ...} tool result fed back to the model, so it can route around a bad call instead of taking the whole run down.
• A model that never converges still returns something. If it exhausts its step/retry budget without valid output, the agent falls back to the same deterministic prefilter → shortlist → score logic as pipeline.py, and reports fallback_used: true.
• Transient API errors get their own retry, via tenacity, separate from the schema-retry loop above — a bad connection and a bad answer are different failure modes.

See docs/CODE_WALKTHROUGH.md for the step-by-step loop.

REST + WebSocket layer (api.py)

A FastAPI service wraps the MCP tools, pipeline.py, and agent.py so they're reachable over HTTP instead of only as CLI scripts:

Endpoint	What it does
GET /health	liveness check
GET /jobs, GET /jobs/{id}	metadata list / full detail, same no-description invariant as the MCP tool
GET /stats	corpus stats
POST /rank	run the ranking pipeline (agentic planning loop by default, or the fixed staged pipeline); use_llm: false runs the free deterministic half only
WS /ws/rank	same as POST /rank, but streams one event per agent step as it happens, instead of a single response at the end

One MCP stdio session is opened once at startup and shared behind a lock (api.MCPSession) rather than spawning a subprocess per request — a deliberate simplification over a real connection pool, documented as such in api.py's module docstring, not oversold as an actual distributed system. Each request gets a correlation id (request_id), threaded through logs and every streamed event, so a run can be traced across the async hops.

MCP notes worth knowing cold

• Why it exists: N models × M integrations becomes N + M. One protocol, JSON-RPC 2.0 over stdio or Streamable HTTP.
• Tools vs resources vs prompts: model-controlled / application-controlled / user-controlled. Getting this trio right is a common interview differentiator.
• CallToolResult shape: content (blocks), structured_content (typed, wrapped as {"result": ...} for non-object returns), is_error. See pipeline.call().
• Tool design is API design for a non-human caller. list_jobs and get_job_details are split because that split is what enables staged retrieval. Docstrings are the tool description the model reads — vague docstring, wrong tool choice.
• Batch parameters over scalar ones: get_job_details(job_ids: list[str]) costs one round trip; get_job_detail(job_id: str) costs eight.

Known limits

• Shortlist recall (does stage 1 keep the labelled-relevant jobs that survive prefilter?) needs a real OPENAI_API_KEY to measure — python eval.py --top 8 runs it; not run here for cost reasons.
• The corpus is synthetic. Real postings are messier — HTML, duplicates, stale listings.
• No caching across runs, so repeated invocations pay stage 1 again.

Stage 1 recall — measured, not assumed

data/relevance_labels.json has 20 job IDs a human would call relevant to the candidate, picked with a documented, reproducible rubric (see the file). eval.py checks two things separately:

• prefilter_recall — of the 20 labelled-relevant jobs, how many survive the deterministic prefilter? Free, no API key: python eval.py --prefilter-only → 20/20, recall 1.0. The label rubric is a strict subset of prefilter's own gates, so this confirms prefilter isn't silently dropping the target role, rather than assuming it.
• shortlist_recall — of those, how many also survive the LLM shortlist at the top you actually run with? python eval.py --top 8 — needs OPENAI_API_KEY, a real model call, so it isn't run in this repo; run it yourself when you have a key.

Your TODOs

1. prefilter() — add seniority and location gates. Re-run --benchmark, record the number. Done: 150 → 31 survivors, 10.8× reduction vs. naive.
2. Swap approx_tokens for real tiktoken counting; note how far the ÷4 heuristic was off. Done: heuristic overestimated by 17.4%.
3. Build a 20-job labelled relevance set and measure stage 1 recall. Done for the free half (prefilter_recall = 1.0); the paid half (shortlist_recall) is wired up in eval.py --top 8, run it with your own key.
4. Wire the server into Claude Desktop's MCP config and call it by hand. Config snippet and restart instructions are in docs/OVERVIEW.md — actually registering it happens in your own Claude Desktop app, not something this repo can do for you.

Docs

• docs/OVERVIEW.md — what this project is, what problem it solves and why, architecture, Claude Desktop wiring, local testability, known limitations.
• docs/CODE_WALKTHROUGH.md — every module, function by function.