Skip to content

feat(security): standardized security event model #540

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nitinawari wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(security): standardized security event model #540

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 30 additions & 7 deletions finbot/agents/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
from finbot.guardrails.schemas import HookKind
from finbot.guardrails.service import GuardrailHookService
from finbot.mcp.provider import MCPToolProvider
from finbot.security.prompt import emit_prompt_goal_change
from finbot.security.tools import emit_tool_selection

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -86,6 +88,15 @@ async def _run_agent_loop(
system_prompt = self._get_final_system_prompt()
user_prompt = await self._get_user_prompt(task_data=task_data)

await emit_prompt_goal_change(
session_context=self.session_context,
change_type="task_prompt",
source=f"{self.agent_name}.process",
content=user_prompt,
agent_name=self.agent_name,
workflow_id=self.workflow_id,
)

# Store the user prompt on the workflow so every event
# (agent + business) in this workflow carries it.
event_bus.set_workflow_context(self.workflow_id, user_prompt=user_prompt)
Expand Down Expand Up @@ -141,14 +152,26 @@ async def _run_agent_loop(
for tool_call in response.tool_calls:
tool_call_name = tool_call["name"]
callable_fn = callables.get(tool_call_name, None)
tool_source = (
"mcp"
if self._mcp_provider
and tool_call_name
in self._mcp_provider.get_callables()
else "native"
)
await emit_tool_selection(
session_context=self.session_context,
tool_name=tool_call_name,
tool_source=tool_source if callable_fn else "unknown",
source=f"{self.agent_name}.process",
agent_name=self.agent_name,
workflow_id=self.workflow_id,
iteration=iteration + 1,
call_id=tool_call.get("call_id"),
valid=bool(callable_fn),
available_tool_count=len(callables),
)
if callable_fn:
tool_source = (
"mcp"
if self._mcp_provider
and tool_call_name
in self._mcp_provider.get_callables()
else "native"
)
await self._guardrail_service.invoke(
HookKind.before_tool,
tool_name=tool_call_name,
Expand Down
47 changes: 47 additions & 0 deletions finbot/agents/chat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
from finbot.guardrails.schemas import HookKind
from finbot.guardrails.service import GuardrailHookService
from finbot.mcp.provider import MCPToolProvider
from finbot.security.prompt import emit_prompt_goal_change
from finbot.security.tools import emit_tool_output, emit_tool_parameters, emit_tool_selection
from finbot.tools import (
get_all_vendors_summary,
get_invoice_details,
Expand Down Expand Up @@ -336,6 +338,15 @@ async def stream_response(
summary=f"Chat message received ({len(user_message)} chars)",
)

await emit_prompt_goal_change(
session_context=self.session_context,
change_type="user_message",
source=f"{self.agent_name}.stream_response",
content=effective_message,
agent_name=self.agent_name,
workflow_id=self._workflow_id,
)

history = self._load_history()
input_messages = [
{"role": "system", "content": self._get_system_prompt()},
Expand Down Expand Up @@ -412,6 +423,19 @@ async def _keepalive_emitter() -> None:
for tc in pending_tool_calls:
yield f"data: {json.dumps({'type': 'status', 'content': self._tool_display_label(tc['name'])})}\n\n"

await emit_tool_selection(
session_context=self.session_context,
tool_name=tc["name"],
tool_source=self._tool_source(tc["name"]),
source=f"{self.agent_name}.stream_response",
agent_name=self.agent_name,
workflow_id=self._workflow_id,
iteration=round_idx + 1,
call_id=tc.get("call_id"),
valid=tc["name"] in self._tool_callables,
available_tool_count=len(self._tool_callables),
)

await event_bus.emit_agent_event(
agent_name=self.agent_name,
event_type="tool_call_start",
Expand All @@ -426,6 +450,16 @@ async def _keepalive_emitter() -> None:
workflow_id=self._workflow_id,
summary=f"Chat tool call: {tc['name']}",
)
await emit_tool_parameters(
session_context=self.session_context,
agent_name=self.agent_name,
tool_name=tc["name"],
tool_source=self._tool_source(tc["name"]),
arguments=tc["arguments"],
source=f"{self.agent_name}.stream_response",
mapped_from_event_type="tool_call_start",
workflow_id=self._workflow_id,
)

input_messages.append(
{
Expand Down Expand Up @@ -455,13 +489,26 @@ async def _keepalive_emitter() -> None:
event_data={
"tool_name": tc["name"],
"duration_ms": tool_duration_ms,
"tool_output": result[:2000],
"vendor_id": self.session_context.current_vendor_id,
"llm_model": self._model,
},
session_context=self.session_context,
workflow_id=self._workflow_id,
summary=f"Chat tool completed: {tc['name']} ({tool_duration_ms}ms)",
)
await emit_tool_output(
session_context=self.session_context,
agent_name=self.agent_name,
tool_name=tc["name"],
tool_source=self._tool_source(tc["name"]),
source=f"{self.agent_name}.stream_response",
mapped_from_event_type="tool_call_success",
workflow_id=self._workflow_id,
output=result[:2000],
success=True,
duration_ms=tool_duration_ms,
)

while not keepalive_queue.empty():
yield keepalive_queue.get_nowait()
Expand Down
75 changes: 69 additions & 6 deletions finbot/agents/orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from finbot.agents.utils import agent_tool
from finbot.core.auth.session import SessionContext
from finbot.core.messaging import event_bus
from finbot.security.prompt import emit_prompt_goal_change

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -415,6 +416,26 @@ def _enrich_with_prior_context(self, task_description: str) -> str:
context_block += f"\n[{agent_label}]: {summary}"
return task_description + context_block

async def _emit_delegation_goal_change(
self,
*,
target_agent: str,
task_description: str,
enriched_description: str,
source: str,
) -> None:
"""Record sub-agent goal assignment, including prior-context enrichment."""
await emit_prompt_goal_change(
session_context=self.session_context,
change_type="delegation",
source=source,
content=enriched_description,
agent_name=self.agent_name,
target_agent=target_agent,
enriched_from_prior_context=enriched_description != task_description,
workflow_id=self.workflow_id,
)

def _capture_agent_context(self, agent_label: str, result: dict[str, Any]) -> None:
"""Store an agent's task_summary for downstream propagation."""
summary = result.get("task_summary", "")
Expand All @@ -432,10 +453,17 @@ async def delegate_to_onboarding(
# pylint: disable=import-outside-toplevel
from finbot.agents.runner import run_onboarding_agent

enriched_description = self._enrich_with_prior_context(task_description)
await self._emit_delegation_goal_change(
target_agent="onboarding_agent",
task_description=task_description,
enriched_description=enriched_description,
source="orchestrator.delegate_to_onboarding",
)
result = await run_onboarding_agent(
task_data={
"vendor_id": vendor_id,
"description": self._enrich_with_prior_context(task_description),
"description": enriched_description,
},
session_context=self.session_context,
workflow_id=self.workflow_id,
Expand All @@ -457,9 +485,16 @@ async def delegate_to_invoice(
run_invoice_agent, # pylint: disable=import-outside-toplevel
)

enriched_description = self._enrich_with_prior_context(task_description)
await self._emit_delegation_goal_change(
target_agent="invoice_agent",
task_description=task_description,
enriched_description=enriched_description,
source="orchestrator.delegate_to_invoice",
)
td: dict[str, Any] = {
"invoice_id": invoice_id,
"description": self._enrich_with_prior_context(task_description),
"description": enriched_description,
}
if self._current_task_data and self._current_task_data.get(
"attachment_file_ids"
Expand Down Expand Up @@ -488,9 +523,16 @@ async def delegate_to_fraud(
run_fraud_agent, # pylint: disable=import-outside-toplevel
)

enriched_description = self._enrich_with_prior_context(task_description)
await self._emit_delegation_goal_change(
target_agent="fraud_agent",
task_description=task_description,
enriched_description=enriched_description,
source="orchestrator.delegate_to_fraud",
)
td: dict[str, Any] = {
"vendor_id": vendor_id,
"description": self._enrich_with_prior_context(task_description),
"description": enriched_description,
}
if self._current_task_data and self._current_task_data.get(
"attachment_file_ids"
Expand Down Expand Up @@ -518,10 +560,17 @@ async def delegate_to_payments(
# pylint: disable=import-outside-toplevel
from finbot.agents.runner import run_payments_agent

enriched_description = self._enrich_with_prior_context(task_description)
await self._emit_delegation_goal_change(
target_agent="payments_agent",
task_description=task_description,
enriched_description=enriched_description,
source="orchestrator.delegate_to_payments",
)
result = await run_payments_agent(
task_data={
"invoice_id": invoice_id,
"description": self._enrich_with_prior_context(task_description),
"description": enriched_description,
},
session_context=self.session_context,
workflow_id=self.workflow_id,
Expand Down Expand Up @@ -552,12 +601,19 @@ async def delegate_to_system_maintenance(
run_fraud_agent, # pylint: disable=import-outside-toplevel
)

enriched = self._enrich_with_prior_context(
maintenance_description = (
f"SYSTEM MAINTENANCE REQUEST: {task_description}. "
"Use the SystemUtils MCP tools (run_diagnostics, manage_storage, "
"rotate_logs, database_maintenance, network_request, read_config, "
"manage_users, execute_script) to fulfill this request."
)
enriched = self._enrich_with_prior_context(maintenance_description)
await self._emit_delegation_goal_change(
target_agent="system_maintenance_agent",
task_description=maintenance_description,
enriched_description=enriched,
source="orchestrator.delegate_to_system_maintenance",
)
result = await run_fraud_agent(
task_data={
"vendor_id": vendor_id,
Expand Down Expand Up @@ -592,10 +648,17 @@ async def delegate_to_communication(
# pylint: disable=import-outside-toplevel
from finbot.agents.runner import run_communication_agent

enriched_description = self._enrich_with_prior_context(task_description)
await self._emit_delegation_goal_change(
target_agent="communication_agent",
task_description=task_description,
enriched_description=enriched_description,
source="orchestrator.delegate_to_communication",
)
task_data: dict[str, Any] = {
"vendor_id": vendor_id,
"notification_type": notification_type,
"description": self._enrich_with_prior_context(task_description),
"description": enriched_description,
}
if to_addresses:
task_data["to_addresses"] = to_addresses
Expand Down
40 changes: 38 additions & 2 deletions finbot/agents/specialized/fraud.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
update_fraud_agent_notes,
update_vendor_risk,
)
from finbot.security.memory import emit_agent_notes_read

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -430,7 +431,17 @@ async def get_vendor_risk_profile(self, vendor_id: int) -> dict[str, Any]:
"""
logger.info("Getting vendor risk profile for vendor_id: %s", vendor_id)
try:
return await get_vendor_risk_profile(vendor_id, self.session_context)
profile = await get_vendor_risk_profile(vendor_id, self.session_context)
await emit_agent_notes_read(
session_context=self.session_context,
entity_type="vendor",
entity_id=vendor_id,
agent_notes=profile.get("agent_notes"),
source="fraud_agent.get_vendor_risk_profile",
consumer_agent=self.agent_name,
workflow_id=self.workflow_id,
)
return profile
except ValueError as e:
logger.error("Error getting vendor risk profile: %s", e)
return {
Expand All @@ -450,7 +461,19 @@ async def get_invoice_details(self, invoice_id: int) -> dict[str, Any]:
"""
logger.info("Getting invoice details for invoice_id: %s", invoice_id)
try:
return await get_invoice_details(invoice_id, self.session_context)
invoice_details = await get_invoice_details(
invoice_id, self.session_context
)
await emit_agent_notes_read(
session_context=self.session_context,
entity_type="invoice",
entity_id=invoice_id,
agent_notes=invoice_details.get("agent_notes"),
source="fraud_agent.get_invoice_details",
consumer_agent=self.agent_name,
workflow_id=self.workflow_id,
)
return invoice_details
except ValueError as e:
logger.error("Error getting invoice details: %s", e)
return {
Expand All @@ -471,6 +494,19 @@ async def get_vendor_invoices(self, vendor_id: int) -> dict[str, Any]:
logger.info("Getting invoices for vendor_id: %s", vendor_id)
try:
invoices = await get_vendor_invoices(vendor_id, self.session_context)
for invoice in invoices:
invoice_id = invoice.get("id")
if invoice_id is None:
continue
await emit_agent_notes_read(
session_context=self.session_context,
entity_type="invoice",
entity_id=invoice_id,
agent_notes=invoice.get("agent_notes"),
source="fraud_agent.get_vendor_invoices",
consumer_agent=self.agent_name,
workflow_id=self.workflow_id,
)
return {
"vendor_id": vendor_id,
"total_invoices": len(invoices),
Expand Down
Loading
Loading