Building a General Contractor Agent: Subcontractor Coordination and Project Management

The General Contractor's Coordination Challenge

A general contractor on a commercial build-out might coordinate 15-20 different subcontractors: demolition, framing, electrical, plumbing, HVAC, drywall, painting, flooring, fire protection, and more. Each trade depends on others finishing first, and every schedule change cascades through the entire project. An AI agent that manages this coordination — tracking who needs to be where, when, and ensuring the right trade is scheduled after its prerequisites are complete — transforms the GC's ability to run multiple projects simultaneously.

The core problem is information flow. When the plumber finishes rough-in a day early, the drywall crew could start sooner — but only if someone tells them. The agent is that someone.

Trade Dependency Management

Construction follows a strict sequence. The agent models trade dependencies and determines which subcontractors can be scheduled at any given point.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional

@dataclass
class TradePhase:
    trade: str
    phase: str                    # "rough_in", "finish", "trim"
    dependencies: list[str]       # list of "trade:phase" that must complete first
    estimated_days: int
    subcontractor_id: Optional[str] = None
    scheduled_start: Optional[datetime] = None
    actual_start: Optional[datetime] = None
    actual_end: Optional[datetime] = None
    status: str = "pending"       # pending, scheduled, in_progress, complete, blocked

class TradeCoordinator:
    def __init__(self, phases: list[TradePhase]):
        self.phases = {f"{p.trade}:{p.phase}": p for p in phases}

    def get_ready_trades(self) -> list[TradePhase]:
        ready = []
        for key, phase in self.phases.items():
            if phase.status != "pending":
                continue
            deps_met = all(
                self.phases.get(dep, TradePhase("", "", [], 0)).status == "complete"
                for dep in phase.dependencies
            )
            if deps_met:
                ready.append(phase)
        return ready

    def complete_phase(self, trade: str, phase: str) -> dict:
        key = f"{trade}:{phase}"
        current = self.phases.get(key)
        if not current:
            return {"error": f"Phase {key} not found"}
        current.status = "complete"
        current.actual_end = datetime.now()

        newly_ready = self.get_ready_trades()
        return {
            "completed": key,
            "newly_available": [f"{p.trade}:{p.phase}" for p in newly_ready],
            "notification_targets": [
                {
                    "subcontractor_id": p.subcontractor_id,
                    "trade": p.trade,
                    "phase": p.phase,
                    "message": f"{trade} {phase} is complete. Your work can begin.",
                }
                for p in newly_ready
                if p.subcontractor_id
            ],
        }

# Example: typical commercial build-out sequence
COMMERCIAL_BUILDOUT = [
    TradePhase("demolition", "full", [], 3),
    TradePhase("framing", "rough", ["demolition:full"], 5),
    TradePhase("electrical", "rough_in", ["framing:rough"], 4),
    TradePhase("plumbing", "rough_in", ["framing:rough"], 4),
    TradePhase("hvac", "rough_in", ["framing:rough"], 3),
    TradePhase("inspection", "rough", ["electrical:rough_in", "plumbing:rough_in", "hvac:rough_in"], 1),
    TradePhase("insulation", "install", ["inspection:rough"], 2),
    TradePhase("drywall", "hang", ["insulation:install"], 3),
    TradePhase("drywall", "finish", ["drywall:hang"], 4),
    TradePhase("painting", "prime_paint", ["drywall:finish"], 3),
    TradePhase("flooring", "install", ["painting:prime_paint"], 3),
    TradePhase("electrical", "trim", ["painting:prime_paint"], 2),
    TradePhase("plumbing", "trim", ["painting:prime_paint"], 2),
    TradePhase("hvac", "trim", ["painting:prime_paint"], 1),
]

Budget Tracking Against Estimates

The agent tracks actual costs against the original estimate and flags budget variances in real time.

@dataclass
class BudgetLineItem:
    category: str
    estimated_amount: float
    committed_amount: float = 0.0    # Subcontract value
    spent_amount: float = 0.0        # Invoices paid
    pending_invoices: float = 0.0

    @property
    def variance(self) -> float:
        return self.estimated_amount - (self.spent_amount + self.pending_invoices)

    @property
    def variance_percentage(self) -> float:
        if self.estimated_amount == 0:
            return 0
        return (self.variance / self.estimated_amount) * 100

class BudgetTracker:
    def __init__(self, line_items: list[BudgetLineItem]):
        self.items = {item.category: item for item in line_items}

    def record_expense(self, category: str, amount: float, invoice_id: str) -> dict:
        item = self.items.get(category)
        if not item:
            return {"error": f"Category {category} not in budget"}
        item.spent_amount += amount
        alert = None
        if item.variance_percentage < -5:
            alert = {
                "type": "over_budget",
                "category": category,
                "overage": abs(item.variance),
                "message": f"{category} is {abs(item.variance_percentage):.1f}% over budget",
            }
        return {
            "category": category,
            "invoice_id": invoice_id,
            "amount": amount,
            "remaining_budget": round(item.variance, 2),
            "variance_pct": round(item.variance_percentage, 1),
            "alert": alert,
        }

    def get_budget_summary(self) -> dict:
        total_estimated = sum(i.estimated_amount for i in self.items.values())
        total_spent = sum(i.spent_amount for i in self.items.values())
        total_pending = sum(i.pending_invoices for i in self.items.values())
        over_budget = [
            {"category": cat, "overage": round(abs(item.variance), 2)}
            for cat, item in self.items.items()
            if item.variance < 0
        ]
        return {
            "total_estimated": round(total_estimated, 2),
            "total_spent": round(total_spent, 2),
            "total_pending": round(total_pending, 2),
            "total_remaining": round(total_estimated - total_spent - total_pending, 2),
            "overall_variance_pct": round(
                ((total_estimated - total_spent - total_pending) / total_estimated) * 100, 1
            ),
            "categories_over_budget": over_budget,
        }

Change Order Management

Change orders are inevitable. The agent captures scope changes, calculates cost impact, and manages the approval workflow.

from enum import Enum

class ChangeOrderStatus(Enum):
    DRAFT = "draft"
    SUBMITTED = "submitted"
    APPROVED = "approved"
    REJECTED = "rejected"

@dataclass
class ChangeOrder:
    co_number: int
    description: str
    reason: str
    requested_by: str
    cost_impact: float
    schedule_impact_days: int
    status: ChangeOrderStatus = ChangeOrderStatus.DRAFT
    trades_affected: list[str] = field(default_factory=list)

class ChangeOrderManager:
    def __init__(self, db, budget_tracker: BudgetTracker):
        self.db = db
        self.budget = budget_tracker
        self.next_co_number = 1

    async def create_change_order(
        self, description: str, reason: str, requested_by: str,
        cost_items: list[dict], schedule_impact_days: int,
        trades_affected: list[str],
    ) -> ChangeOrder:
        total_cost = sum(item["amount"] for item in cost_items)
        co = ChangeOrder(
            co_number=self.next_co_number,
            description=description,
            reason=reason,
            requested_by=requested_by,
            cost_impact=total_cost,
            schedule_impact_days=schedule_impact_days,
            trades_affected=trades_affected,
        )
        self.next_co_number += 1

        await self.db.execute(
            """INSERT INTO change_orders
               (co_number, description, reason, requested_by,
                cost_impact, schedule_impact_days, status)
               VALUES ($1, $2, $3, $4, $5, $6, $7)""",
            co.co_number, description, reason, requested_by,
            total_cost, schedule_impact_days, co.status.value,
        )
        return co

    async def approve_change_order(self, co_number: int) -> dict:
        co = await self.db.fetchrow(
            "SELECT * FROM change_orders WHERE co_number = $1", co_number
        )
        if not co:
            return {"error": f"Change order #{co_number} not found"}

        await self.db.execute(
            "UPDATE change_orders SET status = 'approved' WHERE co_number = $1",
            co_number,
        )
        budget_update = self.budget.record_expense(
            "change_orders", co["cost_impact"], f"CO-{co_number}"
        )
        return {
            "co_number": co_number,
            "status": "approved",
            "cost_impact": co["cost_impact"],
            "schedule_impact_days": co["schedule_impact_days"],
            "budget_update": budget_update,
        }

FAQ

How does the agent handle trades that can work in parallel?

The dependency graph identifies which trades are independent of each other. After framing rough-in is complete, electrical, plumbing, and HVAC rough-in can all proceed simultaneously. The agent recognizes this and sends scheduling notifications to all three subcontractors at once, along with space-sharing coordination to prevent conflicts (e.g., plumber gets kitchen first while electrician starts in bedrooms).

What happens when a subcontractor no-shows?

The agent detects the no-show when the expected check-in does not occur by the scheduled start time. It immediately alerts the GC, calculates the schedule impact, and queries the approved subcontractor list for available replacements with the same trade license. It provides the GC with options ranked by availability and past reliability rating.

How does the change order process prevent scope creep?

Every change order goes through a formal workflow: draft, submit with cost and schedule impact, approve or reject. The agent enforces this by requiring cost justification and trade impact analysis before submission. It also maintains a running total of all approved change orders against the original contract value, giving the GC and owner clear visibility into how changes are affecting the total project cost.

#GeneralContractor #SubcontractorCoordination #ProjectManagement #BudgetTracking #ChangeOrders #AgenticAI #LearnAI #AIEngineering