justin/new-site
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add MCS-150 and BOC-3 service handlers for trucking compliance

Browse source 
MCS-150 Biennial Update ($79):
- Admin-assisted filing (FMCSA Portal requires Login.gov MFA)
- Creates admin todo with intake data and filing steps
- Checks current MCS-150 status via FMCSA API
- Sends status email to client

BOC-3 Process Agent Filing ($149):
- Partners with blanket process agent (NWRA or similar)
- Collects carrier info, submits designation to partner
- Partner files electronically with FMCSA
- Stub for future process agent API integration
- Sends status/confirmation emails

Both follow the same handler pattern as FCC services (admin todo
with structured data when full automation isn't possible).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
justin 2026-05-28 22:37:47 -05:00
parent 79ebcc001a
commit 8441c6f0c0
2 changed files with 462 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

264
scripts/workers/services/boc3_filing.py Normal file
View file

@ -0,0 +1,264 @@
"""
BOC-3 Process Agent Designation Service Handler.
The BOC-3 designates a process agent in every US state who can accept
legal documents on behalf of a motor carrier. The process agent (not
the carrier) files the form with FMCSA.
Model: Performance West partners with a blanket process agent service
(e.g., NWRA or similar) who covers all 48 contiguous states + DC.
We collect the carrier's info, submit the designation to our process
agent partner, they file the BOC-3 electronically with FMCSA.
Service slug: boc3-filing
Price: $149
Gov fee: $0
Intake data needed:
- DOT number
- MC/FF/MX number (docket number)
- Legal name
- DBA name (if any)
- Business address
- Phone number
- Email
- Entity type (carrier, broker, freight forwarder)
Filing flow:
1. Client orders BOC-3 filing
2. We collect intake data
3. We submit designation request to process agent partner
4. Process agent files BOC-3 electronically with FMCSA
5. We verify filing on FMCSA L&I system
6. We send confirmation to client
"""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime
LOG = logging.getLogger("workers.services.boc3_filing")
# Process agent partner details — update when partnership is established
PROCESS_AGENT_PARTNER = {
"name": "TBD — NWRA or similar",
"contact_email": "",
"api_endpoint": None, # Will be set when partner API is available
}
class BOC3FilingHandler:
"""Handle BOC-3 process agent designation orders."""
SERVICE_SLUG = "boc3-filing"
SERVICE_NAME = "BOC-3 Process Agent Filing"
def handle(self, order_data: dict, order_number: str) -> list[str]:
"""
Process a BOC-3 filing order.
Currently creates an admin todo. When process agent partner API
is available, this will automate the submission.
"""
LOG.info("[%s] Processing BOC-3 filing order", order_number)
intake = order_data.get("intake_data") or {}
if isinstance(intake, str):
intake = json.loads(intake)
dot_number = intake.get("dot_number", "")
docket_number = intake.get("docket_number", "") # MC-XXXXXX
entity_name = intake.get("entity_name", order_data.get("customer_name", ""))
customer_email = order_data.get("customer_email", "")
entity_type = intake.get("entity_type", "carrier") # carrier, broker, freight_forwarder
if not dot_number:
LOG.error("[%s] Missing DOT number", order_number)
return []
# Check current BOC-3 status
boc3_status = self._check_boc3_status(dot_number)
# Build the designation request
designation = {
"dot_number": dot_number,
"docket_number": docket_number,
"legal_name": entity_name,
"dba_name": intake.get("dba_name", ""),
"business_address": {
"street": intake.get("address_street", ""),
"city": intake.get("address_city", ""),
"state": intake.get("address_state", ""),
"zip": intake.get("address_zip", ""),
},
"phone": intake.get("phone", ""),
"email": customer_email,
"entity_type": entity_type,
"requested_at": datetime.utcnow().isoformat(),
}
# If we have a process agent API, submit directly
if PROCESS_AGENT_PARTNER.get("api_endpoint"):
success = self._submit_to_process_agent(designation)
if success:
self._send_confirmation_email(order_number, entity_name, dot_number, customer_email)
return []
# Otherwise create admin todo
todo_data = {
"order_number": order_number,
"service": self.SERVICE_NAME,
"designation": designation,
"current_boc3_status": boc3_status,
"steps": [
"1. Submit BOC-3 designation request to process agent partner",
f" Partner: {PROCESS_AGENT_PARTNER['name']}",
f" Email: {PROCESS_AGENT_PARTNER.get('contact_email', 'TBD')}",
"2. Include: DOT#, MC#, legal name, address for all 48 states + DC",
"3. Process agent files BOC-3 electronically with FMCSA",
"4. Verify filing at https://li-public.fmcsa.dot.gov/LIVIEW/pkg_carrquery.prc_carrlist",
"5. Send confirmation to client",
],
}
try:
import psycopg2
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
with conn.cursor() as cur:
cur.execute("""
INSERT INTO admin_todos (
title, category, priority, order_number, service_slug,
description, data, status
) VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending')
""", (
f"BOC-3 Filing — {entity_name} (DOT {dot_number})",
"filing",
"high",
order_number,
self.SERVICE_SLUG,
f"File BOC-3 process agent designation for {entity_name}.\n"
f"DOT: {dot_number}\n"
f"MC/Docket: {docket_number}\n"
f"Type: {entity_type}\n"
f"Customer: {customer_email}\n\n"
f"Submit to process agent partner for electronic filing with FMCSA.",
json.dumps(todo_data),
))
conn.commit()
conn.close()
LOG.info("[%s] Admin todo created for BOC-3 filing", order_number)
except Exception as exc:
LOG.error("[%s] Failed to create admin todo: %s", order_number, exc)
# Send status email
self._send_status_email(order_number, entity_name, dot_number, customer_email)
return []
def _check_boc3_status(self, dot_number: str) -> str:
"""Check if carrier has a BOC-3 on file via FMCSA API."""
try:
import urllib.request
api_key = os.environ.get("FMCSA_API_KEY", "")
if not api_key:
return "API key not configured"
url = (
f"https://mobile.fmcsa.dot.gov/qc/services/carriers/"
f"{dot_number}?webKey={api_key}"
)
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
carrier = data.get("content", {}).get("carrier", {})
# BOC-3 status isn't directly in the API, but we can check
# if authority is active (requires BOC-3 + insurance on file)
common = carrier.get("commonAuthorityStatus", "N")
contract = carrier.get("contractAuthorityStatus", "N")
broker = carrier.get("brokerAuthorityStatus", "N")
if common == "A" or contract == "A" or broker == "A":
return "Authority active (BOC-3 likely on file)"
else:
return "No active authority (BOC-3 may be needed)"
except Exception as exc:
return f"Could not check: {exc}"
def _submit_to_process_agent(self, designation: dict) -> bool:
"""Submit designation to process agent partner via API."""
# TODO: Implement when partner API is available
LOG.warning("Process agent API not configured — falling back to admin todo")
return False
def _send_status_email(self, order_number, entity_name, dot_number, customer_email):
"""Send client an email that we're working on their BOC-3."""
if not customer_email:
return
try:
import smtplib
from email.mime.text import MIMEText
body = (
f"Hi,\n\n"
f"We've received your BOC-3 process agent designation order for "
f"{entity_name} (DOT# {dot_number}).\n\n"
f"Order: {order_number}\n\n"
f"We're submitting your designation to our blanket process agent "
f"who covers all 48 contiguous states plus DC. Once filed with "
f"FMCSA, your operating authority will reflect the active BOC-3.\n\n"
f"This is typically completed within 1-2 business days.\n\n"
f"Questions? Reply to this email or call (888) 411-0383.\n\n"
f"Performance West Inc.\n"
f"DOT Compliance Services\n"
)
msg = MIMEText(body)
msg["Subject"] = f"BOC-3 Filing In Progress — {entity_name} (DOT {dot_number})"
msg["From"] = "noreply@performancewest.net"
msg["To"] = customer_email
with smtplib.SMTP("localhost", 25) as s:
s.sendmail(msg["From"], [customer_email], msg.as_string())
LOG.info("[%s] Status email sent to %s", order_number, customer_email)
except Exception as exc:
LOG.warning("[%s] Failed to send status email: %s", order_number, exc)
def _send_confirmation_email(self, order_number, entity_name, dot_number, customer_email):
"""Send confirmation that BOC-3 has been filed."""
if not customer_email:
return
try:
import smtplib
from email.mime.text import MIMEText
body = (
f"Hi,\n\n"
f"Your BOC-3 process agent designation has been filed with FMCSA "
f"for {entity_name} (DOT# {dot_number}).\n\n"
f"Order: {order_number}\n\n"
f"Your process agent is now designated in all 48 contiguous states "
f"plus the District of Columbia. This designation remains active "
f"as long as your carrier account is maintained.\n\n"
f"You can verify your BOC-3 status at:\n"
f"https://li-public.fmcsa.dot.gov/LIVIEW/pkg_carrquery.prc_carrlist\n\n"
f"Questions? Reply to this email or call (888) 411-0383.\n\n"
f"Performance West Inc.\n"
f"DOT Compliance Services\n"
)
msg = MIMEText(body)
msg["Subject"] = f"BOC-3 Filed — {entity_name} (DOT {dot_number})"
msg["From"] = "noreply@performancewest.net"
msg["To"] = customer_email
with smtplib.SMTP("localhost", 25) as s:
s.sendmail(msg["From"], [customer_email], msg.as_string())
LOG.info("[%s] Confirmation email sent to %s", order_number, customer_email)
except Exception as exc:
LOG.warning("[%s] Failed to send confirmation email: %s", order_number, exc)

198
scripts/workers/services/mcs150_update.py Normal file
View file

@ -0,0 +1,198 @@
"""
MCS-150 Biennial Update Service Handler.
The MCS-150 is filed through the FMCSA Portal (portal.fmcsa.dot.gov) using
Login.gov credentials + MFA. Since we can't automate through MFA, this is
an admin-assisted service:
1. Client orders MCS-150 update
2. We collect their updated info via intake form
3. Admin logs into FMCSA Portal with client's credentials (provided by client)
OR we prepare the data and walk the client through filing via screen share
4. We verify the update was accepted
5. We send confirmation with updated company snapshot
Service slug: mcs150-update
Price: $79
Gov fee: $0
Intake data needed:
- DOT number
- Legal name (confirm/update)
- DBA name (confirm/update)
- Principal business address
- Mailing address
- Phone number
- Email address
- Number of power units
- Number of drivers
- Operation type (interstate/intrastate)
- Carrier operation (authorized for hire, exempt for hire, private)
- Cargo types
- Hazmat (Y/N)
- Annual mileage + year
- FMCSA Portal login credentials (Login.gov email + password)
OR "I need help creating my Login.gov account"
Filing approach:
Option A: Client provides Login.gov credentials → admin files directly
Option B: Guided filing via screen share (Zoom/Teams) → $29 upcharge
Option C: We prepare a pre-filled PDF → client uploads themselves (cheapest)
"""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime
LOG = logging.getLogger("workers.services.mcs150_update")
class MCS150UpdateHandler:
"""Handle MCS-150 biennial update orders."""
SERVICE_SLUG = "mcs150-update"
SERVICE_NAME = "MCS-150 Biennial Update"
def handle(self, order_data: dict, order_number: str) -> list[str]:
"""
Process an MCS-150 update order.
Since FMCSA Portal requires Login.gov MFA, this creates an admin
todo rather than automating the filing directly.
"""
LOG.info("[%s] Processing MCS-150 update order", order_number)
intake = order_data.get("intake_data") or {}
if isinstance(intake, str):
intake = json.loads(intake)
dot_number = intake.get("dot_number", "")
entity_name = intake.get("entity_name", order_data.get("customer_name", ""))
customer_email = order_data.get("customer_email", "")
# Validate required fields
if not dot_number:
LOG.error("[%s] Missing DOT number in intake data", order_number)
return []
# Check current MCS-150 status via FMCSA API
mcs150_status = self._check_current_status(dot_number)
# Create admin todo with all the info needed to file
todo_data = {
"order_number": order_number,
"service": self.SERVICE_NAME,
"dot_number": dot_number,
"entity_name": entity_name,
"customer_email": customer_email,
"current_status": mcs150_status,
"intake_data": intake,
"filing_url": "https://portal.fmcsa.dot.gov/login",
"steps": [
"1. Log into FMCSA Portal with client's Login.gov credentials",
"2. Navigate to Registration > MCS-150",
"3. Update fields with intake data provided",
"4. Verify all information is correct",
"5. Submit the update",
"6. Take screenshot of confirmation",
"7. Download updated company snapshot from SAFER",
"8. Email confirmation + snapshot to client",
],
}
# Create admin todo
try:
import psycopg2
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
with conn.cursor() as cur:
cur.execute("""
INSERT INTO admin_todos (
title, category, priority, order_number, service_slug,
description, data, status
) VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending')
""", (
f"MCS-150 Update — {entity_name} (DOT {dot_number})",
"filing",
"normal",
order_number,
self.SERVICE_SLUG,
f"File MCS-150 biennial update for {entity_name}.\n"
f"DOT: {dot_number}\n"
f"Customer: {customer_email}\n"
f"Current MCS-150 status: {mcs150_status}\n\n"
f"Client intake data attached. Log into FMCSA Portal and update.",
json.dumps(todo_data),
))
conn.commit()
conn.close()
LOG.info("[%s] Admin todo created for MCS-150 update", order_number)
except Exception as exc:
LOG.error("[%s] Failed to create admin todo: %s", order_number, exc)
# Send client a status email
self._send_status_email(order_number, entity_name, dot_number, customer_email)
return [] # No generated files — admin handles the filing
def _check_current_status(self, dot_number: str) -> str:
"""Check current MCS-150 status via FMCSA API."""
try:
import urllib.request
api_key = os.environ.get("FMCSA_API_KEY", "")
if not api_key:
return "API key not configured"
url = f"https://mobile.fmcsa.dot.gov/qc/services/carriers/{dot_number}?webKey={api_key}"
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
carrier = data.get("content", {}).get("carrier", {})
outdated = carrier.get("mcs150Outdated", "?")
status = carrier.get("statusCode", "?")
allowed = carrier.get("allowedToOperate", "?")
return (
f"Status: {status}, Allowed: {allowed}, "
f"MCS-150 Outdated: {outdated}"
)
except Exception as exc:
return f"Could not check: {exc}"
def _send_status_email(self, order_number, entity_name, dot_number, customer_email):
"""Send client an email that we're working on their update."""
if not customer_email:
return
try:
import smtplib
from email.mime.text import MIMEText
body = (
f"Hi,\n\n"
f"We've received your MCS-150 biennial update order for "
f"{entity_name} (DOT# {dot_number}).\n\n"
f"Order: {order_number}\n\n"
f"Our team will review your intake information and complete "
f"the filing within 1-2 business days. We'll send you a "
f"confirmation with your updated company snapshot once it's done.\n\n"
f"If we need your FMCSA Portal login credentials, we'll reach "
f"out via a separate secure email.\n\n"
f"Questions? Reply to this email or call (888) 411-0383.\n\n"
f"Performance West Inc.\n"
f"DOT Compliance Services\n"
)
msg = MIMEText(body)
msg["Subject"] = f"MCS-150 Update In Progress — {entity_name} (DOT {dot_number})"
msg["From"] = "noreply@performancewest.net"
msg["To"] = customer_email
with smtplib.SMTP("localhost", 25) as s:
s.sendmail(msg["From"], [customer_email], msg.as_string())
LOG.info("[%s] Status email sent to %s", order_number, customer_email)
except Exception as exc:
LOG.warning("[%s] Failed to send status email: %s", order_number, exc)