Match experts to project roles from their profiles with Mistral Apps

graph LR
    Page["Web page at /api"] -->|uploads a profile| API["API: /api/v1/experts"]
    API -->|stores the PDF| Files["Files API"]
    API -->|starts| Workflow["match-expert workflow"]
    Workflow -->|reads the profile| OCR["mistral-ocr-latest<br/>with a document annotation"]
    Workflow -->|picks a role| Model["mistral-medium-latest"]
    API -->|roles, experts, and reviews| DB["Managed Postgres"]

mistral login
mistral apps init expert-matching-workflow --caps fastapi,postgres,workflows,apps
cd expert-matching-workflow

uv sync --all-packages

expert-matching-workflow/
├── apps/
│   ├── api/src/api/routers/api/
│   │   ├── ui/
│   │   │   ├── __init__.py
│   │   │   ├── index.html
│   │   │   └── route.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── experts.py
│   │       └── roles.py
│   └── worker/src/worker/workflows/match_expert.py
└── packages/py/
    ├── db/src/db/
    │   ├── accessors/
    │   │   ├── expert.py
    │   │   └── role.py
    │   ├── migrations/versions/0001_expert_matching.py
    │   └── models/
    │       ├── expert.py
    │       └── role.py
    └── env/src/env/matching.py

MISTRAL_API_KEY=your-mistral-api-key
DEPLOYMENT_NAME=expert-matching-your-name

rm -rf .agents node_modules .nxignore package.json bun.lock bunfig.toml nx.json \
  tools/install.sh tools/db.sh tools/lib.sh \
  apps/api/project.json apps/worker/project.json packages/py/db/project.json

**/.nx
.nx

            hint="no selected capability owns a table; add a model and an Alembic revision for it",

def _find_repo_root() -> Path:
    for parent in Path(__file__).resolve().parents:
        if (parent / "apps.json").is_file() and (parent / "pyproject.toml").is_file():
            return parent
    return Path.cwd()

# Executed directly -> run uv with this environment. Sourced -> only export the environment.

"""Settings for expert matching: the models that read a profile and pick a role."""

from env._base import BaseEnv


class Env(BaseEnv):
    ocr_model: str = "mistral-ocr-latest"
    matching_model: str = "mistral-medium-latest"


env = Env()

OCR_MODEL=mistral-ocr-latest
MATCHING_MODEL=mistral-medium-latest

"""A project role an expert can be matched to."""

from sqlmodel import Field, SQLModel


class Role(SQLModel, table=True):
    __tablename__ = "role"

    id: int | None = Field(default=None, primary_key=True)
    title: str
    project: str
    description: str

"""An uploaded profile PDF, the structured profile read from it, and the role it was matched to."""

from datetime import UTC, datetime

from sqlalchemy import JSON, Column
from sqlmodel import Field, SQLModel


def _now() -> datetime:
    return datetime.now(UTC)


class Expert(SQLModel, table=True):
    __tablename__ = "expert"

    id: int | None = Field(default=None, primary_key=True)
    file_name: str
    file_id: str
    execution_id: str | None = None
    # processing, then matched or failed
    status: str = "processing"
    name: str | None = None
    headline: str | None = None
    years_of_experience: int | None = None
    skills: list[str] = Field(default_factory=list, sa_column=Column(JSON, nullable=False))
    summary: str | None = None
    role_id: int | None = None
    reason: str | None = None
    # pending, approved or declined
    review: str = "pending"
    created_at: datetime = Field(default_factory=_now)

"""Create the role and expert tables, and add three sample roles.

Revision ID: 0001
Revises:
Create Date: 2026-11-05

"""

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0001"
down_revision: str | Sequence[str] | None = None
branch_labels: str | Sequence[str] | None = ("expert_matching",)
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    role = op.create_table(
        "role",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("title", sa.String(), nullable=False),
        sa.Column("project", sa.String(), nullable=False),
        sa.Column("description", sa.String(), nullable=False),
    )
    op.create_table(
        "expert",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("file_name", sa.String(), nullable=False),
        sa.Column("file_id", sa.String(), nullable=False),
        sa.Column("execution_id", sa.String(), nullable=True),
        sa.Column("status", sa.String(), nullable=False),
        sa.Column("name", sa.String(), nullable=True),
        sa.Column("headline", sa.String(), nullable=True),
        sa.Column("years_of_experience", sa.Integer(), nullable=True),
        sa.Column("skills", sa.JSON(), nullable=False),
        sa.Column("summary", sa.String(), nullable=True),
        sa.Column("role_id", sa.Integer(), nullable=True),
        sa.Column("reason", sa.String(), nullable=True),
        sa.Column("review", sa.String(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
    )
    op.bulk_insert(
        role,
        [
            {
                "title": "Data platform engineer",
                "project": "Retail demand forecasting",
                "description": (
                    "Build the ingestion pipelines and feature store behind a weekly demand forecast. "
                    "Python, SQL, Spark or similar, and experience running pipelines in production."
                ),
            },
            {
                "title": "Cloud security architect",
                "project": "Bank cloud migration",
                "description": (
                    "Design identity, network and key management for a bank moving workloads to the cloud. "
                    "Experience with regulated environments and security reviews."
                ),
            },
            {
                "title": "Product designer",
                "project": "Patient booking app",
                "description": (
                    "Lead user research and design the booking flow of a mobile app for clinics. "
                    "Strong portfolio in mobile UX and accessibility."
                ),
            },
        ],
    )


def downgrade() -> None:
    op.drop_table("expert")
    op.drop_table("role")

"""Reads and writes of the ``role`` table."""

from sqlmodel import col, select

from db.engine import get_session_maker
from db.models.role import Role


async def create_role(role: Role) -> Role:
    async with get_session_maker()() as session:
        session.add(role)
        await session.commit()
        await session.refresh(role)
    return role


async def list_roles() -> list[Role]:
    async with get_session_maker()() as session:
        rows = await session.execute(select(Role).order_by(col(Role.id)))
        return list(rows.scalars())

"""Reads and writes of the ``expert`` table."""

from sqlmodel import col, select

from db.engine import get_session_maker
from db.models.expert import Expert


async def save_expert(expert: Expert) -> Expert:
    """Insert a new expert, or write the changes to one read earlier."""
    async with get_session_maker()() as session:
        expert = await session.merge(expert)
        await session.commit()
        await session.refresh(expert)
    return expert


async def get_expert(expert_id: int) -> Expert | None:
    async with get_session_maker()() as session:
        return await session.get(Expert, expert_id)


async def list_experts(limit: int = 50) -> list[Expert]:
    async with get_session_maker()() as session:
        rows = await session.execute(select(Expert).order_by(col(Expert.id).desc()).limit(limit))
        return list(rows.scalars())

import json

from mistralai import workflows
from mistralai.workflows import workflow
from pydantic import BaseModel, Field

with workflow.unsafe.imports_passed_through():
    from env.matching import env
    from mistralai.client.models import (
        ChatCompletionRequest,
        FileChunk,
        OCRRequest,
        ResponseFormat,
        SystemMessage,
        UserMessage,
    )
    from mistralai.extra import response_format_from_pydantic_model
    from mistralai.workflows.plugins.mistralai import chat_parse_to_model, mistralai_ocr

PROMPT = """You staff experts on client projects. Given an expert's profile and a list of open roles,
pick the one role the expert fits best. Explain why in two or three sentences, naming the skills
or experience that match, and anything the expert lacks for the role."""


class RoleBrief(BaseModel):
    id: int
    title: str
    project: str
    description: str


class MatchRequest(BaseModel):
    file_id: str
    roles: list[RoleBrief]


class ExpertProfile(BaseModel):
    name: str = Field(description="The expert's full name")
    headline: str = Field(description="Current or most recent job title, with the employer")
    years_of_experience: int = Field(description="Total years of professional experience")
    skills: list[str] = Field(description="Up to ten key skills, tools or domains")
    summary: str = Field(description="Two sentences on what the expert has done")


class RoleMatch(BaseModel):
    role_id: int
    reason: str


class MatchResult(BaseModel):
    profile: ExpertProfile
    match: RoleMatch


@workflows.workflow.define(
    name="match-expert",
    workflow_display_name="Match expert",
    workflow_description="Read an expert's profile and pick the project role that suits the expert best.",
)
class MatchExpert:
    @workflows.workflow.entrypoint
    async def run(self, request: MatchRequest) -> MatchResult:
        ocr = await mistralai_ocr(
            OCRRequest(
                model=env.ocr_model,
                document=FileChunk(file_id=request.file_id),
                document_annotation_format=ResponseFormat.model_validate(
                    response_format_from_pydantic_model(ExpertProfile)
                ),
            )
        )
        profile = ExpertProfile.model_validate_json(ocr.document_annotation)

        roles = json.dumps([role.model_dump() for role in request.roles], indent=2)
        match = await chat_parse_to_model(
            RoleMatch,
            ChatCompletionRequest(
                model=env.matching_model,
                messages=[
                    SystemMessage(content=PROMPT),
                    UserMessage(content=f"Expert profile:\n{profile.model_dump_json(indent=2)}\n\nRoles:\n{roles}"),
                ],
            ),
        )
        return MatchResult(profile=profile, match=match)

"""The matching API. Every route needs a caller signed in through the Apps gateway."""

from fastapi import HTTPException, Request, params, status
from mistralai_capabilities.apps.tokens import token_from


def require_caller(request: Request) -> None:
    # The gateway signs the caller in and forwards a token; a request without one bypassed it.
    if not token_from(request.headers):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Call this API through the app's gateway")


dependencies: tuple[params.Depends, ...] = (params.Depends(require_caller),)

from db.accessors.role import create_role, list_roles
from db.models.role import Role
from fastapi import APIRouter, status
from pydantic import BaseModel

router = APIRouter(tags=["roles"])


class NewRole(BaseModel):
    title: str
    project: str
    description: str


class RoleOut(NewRole):
    id: int


@router.get("")
async def list_all() -> list[RoleOut]:
    """List the open roles experts are matched against."""
    return [RoleOut.model_validate(r, from_attributes=True) for r in await list_roles()]


@router.post("", status_code=status.HTTP_201_CREATED)
async def create(body: NewRole) -> RoleOut:
    """Add an open role."""
    role = await create_role(Role(**body.model_dump()))
    return RoleOut.model_validate(role, from_attributes=True)

from datetime import datetime
from typing import Literal

from db.accessors.expert import get_expert, list_experts, save_expert
from db.accessors.role import list_roles
from db.models.expert import Expert
from fastapi import APIRouter, HTTPException, UploadFile, status
from mistralai_capabilities.workflows.client import get_client, start_workflow
from pydantic import BaseModel
from utils.mistral import mistral_client
from worker.workflows.match_expert import (
    MatchExpert,
    MatchRequest,
    MatchResult,
    RoleBrief,
)

router = APIRouter(tags=["experts"])

FAILED = {"FAILED", "CANCELED", "TERMINATED", "TIMED_OUT"}


class ExpertOut(BaseModel):
    id: int
    file_name: str
    status: str
    name: str | None
    headline: str | None
    years_of_experience: int | None
    skills: list[str]
    summary: str | None
    role_id: int | None
    reason: str | None
    review: str
    created_at: datetime


class Review(BaseModel):
    review: Literal["approved", "declined"]


async def _refresh(expert: Expert) -> Expert:
    """Store the workflow's result once it has finished."""
    client = await get_client()
    execution = await client.workflows.executions.get_workflow_execution_async(execution_id=expert.execution_id)
    if execution.status == "COMPLETED":
        result = MatchResult.model_validate(execution.result)
        expert.sqlmodel_update(result.profile.model_dump())
        expert.role_id = result.match.role_id
        expert.reason = result.match.reason
        expert.status = "matched"
    elif execution.status in FAILED:
        expert.status = "failed"
    else:
        return expert
    await mistral_client().files.delete_async(file_id=expert.file_id)
    return await save_expert(expert)


@router.post("", status_code=status.HTTP_202_ACCEPTED)
async def upload(file: UploadFile) -> ExpertOut:
    """Upload an expert's profile as a PDF and start matching it to a role."""
    roles = await list_roles()
    if not roles:
        raise HTTPException(status.HTTP_409_CONFLICT, "Add a role first")
    uploaded = await mistral_client().files.upload_async(
        file={"file_name": file.filename or "profile.pdf", "content": await file.read()},
        purpose="ocr",
    )
    run = await start_workflow(
        MatchExpert,
        MatchRequest(
            file_id=uploaded.id,
            roles=[RoleBrief.model_validate(r, from_attributes=True) for r in roles],
        ),
        wait_for_result=False,
    )
    expert = await save_expert(
        Expert(
            file_name=uploaded.filename,
            file_id=uploaded.id,
            execution_id=run.execution_id,
        )
    )
    return ExpertOut.model_validate(expert, from_attributes=True)


@router.get("")
async def list_all(limit: int = 50) -> list[ExpertOut]:
    """List the uploaded experts, newest first, with the role each was matched to."""
    experts = [await _refresh(e) if e.status == "processing" else e for e in await list_experts(limit)]
    return [ExpertOut.model_validate(e, from_attributes=True) for e in experts]


@router.post("/{expert_id}/review")
async def review(expert_id: int, body: Review) -> ExpertOut:
    """Approve or decline the role an expert was matched to."""
    expert = await get_expert(expert_id)
    if expert is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "No such expert")
    if expert.status != "matched":
        raise HTTPException(status.HTTP_409_CONFLICT, "This expert has no match to review yet")
    expert.review = body.review
    return ExpertOut.model_validate(await save_expert(expert), from_attributes=True)

"""The web page. It holds no data, so it is served without a caller; the API calls it makes are not."""

from fastapi import params

# Serve the page at the API prefix itself: ``route.py`` is ``/api``.
segment = ""
dependencies: tuple[params.Depends, ...] = ()

from pathlib import Path

from fastapi import APIRouter
from fastapi.responses import HTMLResponse

router = APIRouter(include_in_schema=False)

PAGE = (Path(__file__).parent / "index.html").read_text(encoding="utf-8")


@router.get("", response_class=HTMLResponse)
async def page() -> str:
    return PAGE

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Expert matching</title>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 960px; margin: 2rem auto; padding: 0 1rem; color: #1a1a1a; }
    h2 { margin-top: 2rem; }
    form { display: flex; gap: .5rem; flex-wrap: wrap; align-items: center; }
    input, textarea, button { font: inherit; padding: .4rem .6rem; }
    textarea { width: 100%; }
    table { width: 100%; border-collapse: collapse; }
    th, td { text-align: left; vertical-align: top; padding: .5rem; border-bottom: 1px solid #ddd; }
    .muted { color: #666; font-size: .9em; }
    .error { color: #b00020; }
  </style>
</head>
<body>
  <h1>Expert matching</h1>
  <p>Upload an expert's profile. The agent reads it and picks the open role that suits the expert best.</p>

  <form id="upload">
    <input type="file" name="file" accept="application/pdf" required>
    <button>Upload profile</button>
    <span id="upload-status" class="muted"></span>
  </form>

  <h2>Experts</h2>
  <table>
    <thead><tr><th>Expert</th><th>Best role</th><th>Review</th></tr></thead>
    <tbody id="experts"></tbody>
  </table>

  <h2>Open roles</h2>
  <ul id="roles"></ul>
  <form id="new-role">
    <input name="title" placeholder="Role title" required>
    <input name="project" placeholder="Project" required>
    <textarea name="description" rows="2" placeholder="What the role needs" required></textarea>
    <button>Add role</button>
  </form>

  <script>
    const api = (path, options) =>
      fetch(`/api/v1/${path}`, options).then(async (response) => {
        if (!response.ok) throw new Error((await response.json()).detail ?? response.statusText);
        return response.json();
      });
    const escape = (text) => String(text ?? "").replace(/[&<>"]/g, (c) => `&#${c.charCodeAt(0)};`);
    let roles = [];

    async function loadRoles() {
      roles = await api("roles");
      document.getElementById("roles").innerHTML = roles
        .map((r) => `<li><strong>${escape(r.title)}</strong>, ${escape(r.project)}<br><span class="muted">${escape(r.description)}</span></li>`)
        .join("");
    }

    function describe(expert) {
      if (expert.status === "processing") return `<td>${escape(expert.file_name)}<br><span class="muted">Reading the profile…</span></td><td></td><td></td>`;
      if (expert.status === "failed") return `<td>${escape(expert.file_name)}<br><span class="error">Matching failed</span></td><td></td><td></td>`;
      const role = roles.find((r) => r.id === expert.role_id);
      const review = expert.review === "pending"
        ? `<button onclick="review(${expert.id}, 'approved')">Approve</button> <button onclick="review(${expert.id}, 'declined')">Decline</button>`
        : escape(expert.review);
      return `<td><strong>${escape(expert.name)}</strong>, ${escape(expert.headline)}<br>
          <span class="muted">${escape(expert.years_of_experience)} years · ${expert.skills.map(escape).join(", ")}</span></td>
        <td><strong>${escape(role?.title ?? `Role ${expert.role_id}`)}</strong><br><span class="muted">${escape(expert.reason)}</span></td>
        <td>${review}</td>`;
    }

    async function loadExperts() {
      const experts = await api("experts");
      document.getElementById("experts").innerHTML = experts.map((e) => `<tr>${describe(e)}</tr>`).join("");
      if (experts.some((e) => e.status === "processing")) setTimeout(loadExperts, 3000);
    }

    async function review(id, decision) {
      await api(`experts/${id}/review`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ review: decision }),
      });
      await loadExperts();
    }

    document.getElementById("upload").addEventListener("submit", async (event) => {
      event.preventDefault();
      const status = document.getElementById("upload-status");
      status.textContent = "Uploading…";
      try {
        await api("experts", { method: "POST", body: new FormData(event.target) });
        status.textContent = "";
        event.target.reset();
        await loadExperts();
      } catch (error) {
        status.textContent = error.message;
      }
    });

    document.getElementById("new-role").addEventListener("submit", async (event) => {
      event.preventDefault();
      await api("roles", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(Object.fromEntries(new FormData(event.target))),
      });
      event.target.reset();
      await loadRoles();
    });

    loadRoles().then(loadExperts);
  </script>
</body>
</html>