Skip to content

Email Colleague

Email Colleague -- mvp.email_colleague

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview

Email-based remote steering of the G6 system. Sends formatted status reports, alerts, and recommendations to human operators; polls Gmail inbox for commands; parses and executes email directives; manages email threads with HITL approval workflows; and reports source/provider capabilities without sending or polling.

Supports 11 operations: outbound messaging (send_status_report, send_alert, send_recommendation), inbound command processing (poll_inbox, parse_reply, execute_command), thread management (list_threads, get_thread, close_thread), applied pattern discovery (list_patterns), and provider readiness discovery (source_capabilities). The ops/help surface returns the same operation list.

Current public fields: approver_id, run_mode, reviewer_signature, agentic_evidence, completion_state, warning_card, evidence, request_id, task_id, and run_id are surfaced by the schema and MCP serialization.

When to use:

  • Notifying operators of incident resolution or system status changes
  • Enabling remote command-and-control via email for air-gapped or low-bandwidth environments
  • Threading multi-step approval workflows where Slack/chat is unavailable
  • Checking Gmail, Resend, SQLite, LLM triage, and command-execution readiness before attempting work

Example:

from mvp.email_colleague import EmailColleagueBlock, EmailColleagueInput

block = EmailColleagueBlock()
result = block.infer(EmailColleagueInput(
    op="source_capabilities",
    request_id="req-1",
    task_id="task-1",
    run_id="run-1",
))
# result.value.completion_state -> "qualified-draft" when optional providers are unavailable
# result.value.value["providers"] -> readiness booleans without secret values

Caveats and known limitations:

  • Requires Gmail OAuth setup; credentials must be configured externally before live polling
  • Requires Resend API configuration for actual outbound delivery; missing credentials are reported as qualified-draft degradation
  • Thread storage defaults to SQLite in-memory; durable deployments should pass a persistent db_path
  • Email command parser is deterministic-first; optional LLM triage can only resolve benign natural-language gaps and cannot clear deterministic review flags
  • execute_command requires an explicit approver_id; parsed email commands can never self-authorize
  • Production remote-authority readiness is beta/qualified until identity assurance, spoof/replay policy, encryption policy, delivery guarantees, and expert sign-off policy exist
  • No end-to-end encryption for stored or transmitted emails is declared

Works well with: work_loop, task_tracker, diagnostic_collector

Public API

EmailColleagueBlock(AIBlock[EmailColleagueInput, EmailColleagueOutput, None])

Email-based remote steering for G6 agents.

Field Type Default
name str 'email_colleague'
db_path str ':memory:'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)

Methods:

infer(data: EmailColleagueInput) -> Result[EmailColleagueOutput]

GmailPoller

Polls a Gmail inbox for new messages.

Constructor:

Parameter Type Default
credentials_path str _DEFAULT_CREDS
is_processed Callable[[str], bool] \| None None
mark_processed Callable[[str], None] \| None None

Methods:

last_poll_time() -> str

processed_count() -> int

poll(max_results: int = 10) -> list[dict[str, Any]]

Fetch new unread messages from Gmail inbox.

mark_processed(message_id: str) -> None

reset() -> None

EmailDirection(str, Enum)

EmailCommandType(str, Enum)

EmailThread(BaseModel)

Field Type Default
thread_id str required
subject str required
participants list[str] Field(default_factory=list)
status Literal['open', 'closed'] 'open'
created_at str ''
updated_at str ''
message_count int 0
metadata dict[str, Any] Field(default_factory=dict)

EmailMessage(BaseModel)

Field Type Default
message_id str required
thread_id str required
direction EmailDirection required
sender str required
recipients list[str] Field(default_factory=list)
subject str ''
body_text str ''
body_html str ''
timestamp str ''
command_type EmailCommandType \| None None
command_params dict[str, Any] Field(default_factory=dict)

EmailColleagueInput(BaseModel)

Field Type Default
op _OPS required
thread_id str ''
to str ''
subject str ''
body_text str ''
body_html str ''
data dict[str, Any] Field(default_factory=dict)
command_type EmailCommandType \| None None
command_params dict[str, Any] Field(default_factory=dict)
approver_id str ''
limit int 50
run_mode str 'beta'
reviewer_signature str ''
dry_run bool True
approval_confirmed bool False
request_id str ''
task_id str ''
run_id str ''

EmailColleagueOutput(BaseModel)

Field Type Default
op str required
success bool True
thread_id str ''
message_id str ''
thread EmailThread \| None None
threads list[EmailThread] Field(default_factory=list)
messages list[EmailMessage] Field(default_factory=list)
command_type EmailCommandType \| None None
command_params dict[str, Any] Field(default_factory=dict)
count int 0
value Any None
error str ''
degraded bool False
degradation_reason str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] Field(default_factory=dict)
evidence dict[str, Any] Field(default_factory=dict)
request_id str ''
task_id str ''
run_id str ''
agentic_evidence dict \| None None

EmailColleagueStore

Thread-safe SQLite store for email colleague state.

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

create_thread(subject: str, participants: list[str], metadata: dict | None = None) -> str

get_thread(thread_id: str) -> dict | None

list_threads(status: str | None = None, limit: int = 50) -> list[dict]

close_thread(thread_id: str) -> bool

add_message(thread_id: str, direction: str, sender: str, recipients: list[str], subject: str = '', body_text: str = '', body_html: str = '', command_type: str | None = None, command_params: dict | None = None) -> str

get_thread_messages(thread_id: str, limit: int = 100) -> list[dict]

log_command(message_id: str, thread_id: str, command_type: str, params: dict | None = None) -> str

mark_command_executed(command_id: str, result: dict | None = None) -> None

mark_processed(message_id: str) -> None

Durably record a Gmail message ID as processed.

is_processed(message_id: str) -> bool

Return True if this Gmail message ID was already processed.

list_processed_ids(limit: int = 1000) -> list[str]

Return processed Gmail message IDs (most recent first).

ThreadManager

Manages conversational email threads.

Constructor:

Parameter Type Default
store EmailColleagueStore required

Methods:

create_thread(subject: str, participants: list[str], metadata: dict | None = None) -> EmailThread

add_message(thread_id: str, direction: EmailDirection, sender: str, recipients: list[str], subject: str = '', body_text: str = '', body_html: str = '', command_type: str | None = None, command_params: dict | None = None) -> str

get_thread_messages(thread_id: str, limit: int = 100) -> list[EmailMessage]

close_thread(thread_id: str) -> bool

get_thread(thread_id: str) -> EmailThread | None

list_threads(status: str | None = None, limit: int = 50) -> list[EmailThread]

Functions

format_status_report(data: dict[str, Any]) -> str

format_alert(data: dict[str, Any]) -> str

format_recommendation(data: dict[str, Any]) -> str

parse_email_command(body_text: str) -> tuple[EmailCommandType | None, dict[str, Any]]

Parse a structured command from email body text.

send_email(to: str | list[str], subject: str, body_html: str, reply_to_thread_id: str | None = None, from_address: str = _FROM_ADDRESS, dry_run: bool = True, approval_confirmed: bool = False) -> dict[str, Any]

Send an email via Resend.