External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

++

Browse Source
This commit is contained in:
Renn F
2025-12-13 13:25:37 +01:00
parent 9e7d9e81e2
commit 4f1d59987f
20 changed files with 3043 additions and 2295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
roboco/mcp/journal_server.py
+380 -479
View File
@@ -19,6 +19,7 @@ from typing import Any
import httpx
from fastapi import status
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from roboco.config import settings
from roboco.llm import ToonAdapter
@@ -26,6 +27,99 @@ from roboco.llm import ToonAdapter
# Global TOON adapter for encoding journal data
_toon = ToonAdapter()
# =============================================================================
# INPUT MODELS (Pydantic models to reduce argument count)
# =============================================================================
class JournalEntryInput(BaseModel):
"""Input for creating a general journal entry."""
title: str = Field(..., description="Entry title (short description)")
content: str = Field(..., description="Entry content (detailed text)")
entry_type: str = Field(
default="general",
description="Type: general, task_reflection, decision_log, learning, struggle",
)
task_id: str | None = Field(default=None, description="Optional related task")
tags: list[str] = Field(default_factory=list, description="Optional list of tags")
is_private: bool = Field(
default=False, description="If true, only you and CEO/Auditor can see"
)
class TaskReflectionInput(BaseModel):
"""Input for creating a task reflection entry."""
task_id: str = Field(..., description="The task UUID you're reflecting on")
title: str = Field(..., description="Reflection title")
what_done: str = Field(..., description="What was accomplished")
what_learned: str = Field(..., description="Key learnings from this task")
what_struggled: str = Field(..., description="What was difficult or challenging")
next_steps: list[str] = Field(
default_factory=list, description="Optional follow-up items"
)
tags: list[str] = Field(default_factory=list, description="Optional list of tags")
class DecisionOption(BaseModel):
"""A decision option with pros/cons."""
option: str
pros_cons: str
class DecisionLogInput(BaseModel):
"""Input for logging a decision."""
title: str = Field(..., description="Decision title")
context: str = Field(..., description="What situation led to this decision")
options: list[DecisionOption] = Field(
..., min_length=2, description="Options considered (at least 2)"
)
chosen: str = Field(..., description="Which option was chosen")
rationale: str = Field(..., description="Why this option was chosen")
consequences: list[str] = Field(
default_factory=list, description="Expected consequences"
)
task_id: str | None = Field(default=None, description="Optional related task")
tags: list[str] = Field(default_factory=list, description="Optional list of tags")
class LearningInput(BaseModel):
"""Input for logging a learning."""
title: str = Field(..., description="Learning title")
what_learned: str = Field(..., description="The actual learning/insight")
how_applied: str | None = Field(
default=None, description="How you applied or plan to apply this"
)
source: str | None = Field(
default=None, description="Where you learned this (docs, experiment, etc.)"
)
task_id: str | None = Field(default=None, description="Optional related task")
tags: list[str] = Field(default_factory=list, description="Optional list of tags")
class StruggleInput(BaseModel):
"""Input for logging a struggle."""
title: str = Field(..., description="Struggle title")
what_struggled: str = Field(..., description="What the challenge was")
attempted_solutions: list[str] = Field(
default_factory=list, description="What you tried (even if it didn't work)"
)
resolution: str | None = Field(
default=None, description="How it was resolved (if resolved)"
)
help_needed: str | None = Field(
default=None, description="What help you need (if unresolved)"
)
task_id: str | None = Field(default=None, description="Optional related task")
tags: list[str] = Field(default_factory=list, description="Optional list of tags")
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
@@ -51,6 +145,262 @@ def _format_error_response(
}
async def _post_journal_entry(
endpoint: str,
payload: dict[str, Any],
agent_id: str,
) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
"""Post to a journal endpoint. Returns (data, error)."""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{_get_api_url()}/journals/me/{endpoint}",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return None, _format_error_response(
"CREATE_FAILED",
f"Failed to create {endpoint.rstrip('s')}",
{"api_error": resp.text},
)
return resp.json(), None
# =============================================================================
# TOOL IMPLEMENTATIONS
# =============================================================================
async def _handle_journal_entry(
data: JournalEntryInput, agent_id: str
) -> dict[str, Any]:
"""Handle journal entry creation."""
valid_types = ["general", "task_reflection", "decision_log", "learning", "struggle"]
if data.entry_type not in valid_types:
return _format_error_response(
"INVALID_TYPE",
f"Invalid entry type. Must be one of: {valid_types}",
)
payload = {
"type": data.entry_type,
"title": data.title,
"content": data.content,
"task_id": data.task_id,
"tags": data.tags,
"is_private": data.is_private,
}
entry, error = await _post_journal_entry("entries", payload, agent_id)
if error or entry is None:
return error or _format_error_response("ERROR", "Failed to create entry")
return {
"status": "created",
"entry": entry,
"entry_toon": _toon.encode(entry),
"guidance": (
"Journal entry saved. Use roboco_journal_search to find past entries."
),
}
async def _handle_reflect(data: TaskReflectionInput, agent_id: str) -> dict[str, Any]:
"""Handle task reflection creation."""
payload = {
"task_id": data.task_id,
"title": data.title,
"what_done": data.what_done,
"what_learned": data.what_learned,
"what_struggled": data.what_struggled,
"next_steps": data.next_steps,
"tags": data.tags,
}
entry, error = await _post_journal_entry("reflections", payload, agent_id)
if error:
return error
return {
"status": "created",
"entry": entry,
"guidance": (
"Reflection saved. This will help you (and future you) "
"when working on similar tasks."
),
}
async def _handle_decision(data: DecisionLogInput, agent_id: str) -> dict[str, Any]:
"""Handle decision log creation."""
payload = {
"title": data.title,
"context": data.context,
"options": [opt.model_dump() for opt in data.options],
"chosen": data.chosen,
"rationale": data.rationale,
"consequences": data.consequences,
"task_id": data.task_id,
"tags": data.tags,
}
entry, error = await _post_journal_entry("decisions", payload, agent_id)
if error:
return error
return {
"status": "created",
"entry": entry,
"guidance": (
"Decision logged. If you need to revisit this decision later, "
"you'll have the context of why it was made."
),
}
async def _handle_learning(data: LearningInput, agent_id: str) -> dict[str, Any]:
"""Handle learning entry creation."""
payload = {
"title": data.title,
"what_learned": data.what_learned,
"how_applied": data.how_applied,
"source": data.source,
"task_id": data.task_id,
"tags": data.tags,
}
entry, error = await _post_journal_entry("learnings", payload, agent_id)
if error:
return error
return {
"status": "created",
"entry": entry,
"guidance": "Learning recorded. Use tags to make it searchable later.",
}
async def _handle_struggle(data: StruggleInput, agent_id: str) -> dict[str, Any]:
"""Handle struggle entry creation."""
payload = {
"title": data.title,
"what_struggled": data.what_struggled,
"attempted_solutions": data.attempted_solutions,
"resolution": data.resolution,
"help_needed": data.help_needed,
"task_id": data.task_id,
"tags": data.tags,
}
entry, error = await _post_journal_entry("struggles", payload, agent_id)
if error:
return error
guidance = "Struggle recorded."
if data.help_needed and not data.resolution:
guidance += (
" Since you indicated help is needed, consider asking in your cell channel."
)
return {"status": "created", "entry": entry, "guidance": guidance}
async def _handle_search(query: str, top_k: int, agent_id: str) -> dict[str, Any]:
"""Handle journal search."""
async with httpx.AsyncClient() as client:
payload = {"query": query, "top_k": min(top_k, 20)}
resp = await client.post(
f"{_get_api_url()}/journals/me/search",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"SEARCH_FAILED", "Failed to search journal", {"api_error": resp.text}
)
entries = resp.json()
if not entries:
return {
"entries": [],
"guidance": "No matching entries found. Try different keywords.",
}
return {
"entries": entries,
"count": len(entries),
"guidance": f"Found {len(entries)} relevant entries.",
}
async def _handle_stats(agent_id: str) -> dict[str, Any]:
"""Handle journal stats retrieval."""
async with httpx.AsyncClient() as client:
stats_resp = await client.get(
f"{_get_api_url()}/journals/me/stats",
headers={"X-Agent-Id": agent_id},
)
growth_resp = await client.get(
f"{_get_api_url()}/journals/me/growth",
headers={"X-Agent-Id": agent_id},
)
stats = (
stats_resp.json() if stats_resp.status_code == status.HTTP_200_OK else {}
)
growth = (
growth_resp.json() if growth_resp.status_code == status.HTTP_200_OK else {}
)
return {
"total_entries": stats.get("total_entries", 0),
"entries_by_type": stats.get("entries_by_type", {}),
"last_entry_at": stats.get("last_entry_at"),
"growth_metrics": {
"total_reflections": growth.get("total_reflections", 0),
"total_learnings": growth.get("total_learnings", 0),
"total_struggles": growth.get("total_struggles", 0),
"total_decisions": growth.get("total_decisions", 0),
"struggle_resolution_rate": growth.get("struggle_resolution_rate", 0),
"sentiment_trend": growth.get("sentiment_trend", "stable"),
},
"guidance": (
"These stats reflect your journal activity. "
"Regular journaling helps build context for future sessions."
),
}
async def _handle_recent(
entry_type: str | None,
task_id: str | None,
limit: int,
agent_id: str,
) -> dict[str, Any]:
"""Handle recent entries retrieval."""
async with httpx.AsyncClient() as client:
params: dict[str, Any] = {"limit": min(limit, 50)}
if entry_type:
params["entry_type"] = entry_type
if task_id:
params["task_id"] = task_id
resp = await client.get(
f"{_get_api_url()}/journals/me/entries",
params=params,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response("LIST_FAILED", "Failed to list entries")
entries = resp.json()
return {"entries": entries, "count": len(entries)}
# =============================================================================
# MCP SERVER FACTORY
# =============================================================================
@@ -68,430 +418,63 @@ def create_journal_mcp_server(agent_id: str) -> FastMCP:
"""
mcp = FastMCP(f"roboco-journal-{agent_id}", json_response=True)
# =========================================================================
# GENERAL ENTRY
# =========================================================================
@mcp.tool()
async def roboco_journal_entry(
title: str,
content: str,
entry_type: str = "general",
task_id: str | None = None,
tags: list[str] | None = None,
is_private: bool = False,
) -> dict[str, Any]:
async def roboco_journal_entry(data: JournalEntryInput) -> dict[str, Any]:
"""
Create a general journal entry.
Your journal is personal - use it to:
- Track your thoughts and progress
- Record context for future sessions
- Document your journey on tasks
- Note things you've learned or struggled with
Args:
title: Entry title (short description)
content: Entry content (detailed text)
entry_type: Type of entry (general, task_reflection, decision_log, learning, struggle)
task_id: Optional related task
tags: Optional list of tags
is_private: If true, only you and CEO/Auditor can see
Returns:
Created entry
Your journal is personal - use it to track thoughts, progress,
and document your journey on tasks.
"""
valid_types = [
"general",
"task_reflection",
"decision_log",
"learning",
"struggle",
]
if entry_type not in valid_types:
return _format_error_response(
"INVALID_TYPE",
f"Invalid entry type. Must be one of: {valid_types}",
)
async with httpx.AsyncClient() as client:
payload = {
"type": entry_type,
"title": title,
"content": content,
"task_id": task_id,
"tags": tags or [],
"is_private": is_private,
}
resp = await client.post(
f"{_get_api_url()}/journals/me/entries",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return _format_error_response(
"CREATE_FAILED",
"Failed to create journal entry",
{"api_error": resp.text},
)
entry = resp.json()
return {
"status": "created",
"entry": entry,
"entry_toon": _toon.encode(entry), # TOON-encoded for LLM token efficiency
"guidance": "Journal entry saved. Use roboco_journal_search to find past entries.",
}
# =========================================================================
# TASK REFLECTION (Important - called at task completion)
# =========================================================================
return await _handle_journal_entry(data, agent_id)
@mcp.tool()
async def roboco_journal_reflect(
task_id: str,
title: str,
what_done: str,
what_learned: str,
what_struggled: str,
next_steps: list[str] | None = None,
tags: list[str] | None = None,
) -> dict[str, Any]:
async def roboco_journal_reflect(data: TaskReflectionInput) -> dict[str, Any]:
"""
Add a task reflection entry.
IMPORTANT: Call this when completing a task. Reflections help you:
- Build institutional memory
- Track your growth
- Provide context for future similar tasks
Args:
task_id: The task UUID you're reflecting on
title: Reflection title
what_done: What was accomplished
what_learned: Key learnings from this task
what_struggled: What was difficult or challenging
next_steps: Optional list of follow-up items
tags: Optional list of tags
Returns:
Created reflection entry
IMPORTANT: Call this when completing a task. Reflections help build
institutional memory and track your growth.
"""
async with httpx.AsyncClient() as client:
payload = {
"task_id": task_id,
"title": title,
"what_done": what_done,
"what_learned": what_learned,
"what_struggled": what_struggled,
"next_steps": next_steps or [],
"tags": tags or [],
}
resp = await client.post(
f"{_get_api_url()}/journals/me/reflections",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return _format_error_response(
"CREATE_FAILED",
"Failed to create reflection",
{"api_error": resp.text},
)
entry = resp.json()
return {
"status": "created",
"entry": entry,
"guidance": (
"Reflection saved. This will help you (and future you) "
"when working on similar tasks."
),
}
# =========================================================================
# DECISION LOG
# =========================================================================
return await _handle_reflect(data, agent_id)
@mcp.tool()
async def roboco_journal_decision(
title: str,
context: str,
options: list[dict[str, str]],
chosen: str,
rationale: str,
consequences: list[str] | None = None,
task_id: str | None = None,
tags: list[str] | None = None,
) -> dict[str, Any]:
async def roboco_journal_decision(data: DecisionLogInput) -> dict[str, Any]:
"""
Log a decision you made.
Use this when you:
- Choose between multiple approaches
- Make architectural decisions
- Pick one solution over another
This creates a record of WHY you made the decision,
which is valuable for future context.
Args:
title: Decision title
context: What situation led to this decision
options: List of options considered, each with 'option' and 'pros_cons' keys
chosen: Which option was chosen
rationale: Why this option was chosen
consequences: Expected consequences of this decision
task_id: Optional related task
tags: Optional list of tags
Returns:
Created decision log entry
Use when choosing between approaches. Creates a record of WHY
you made the decision for future context.
"""
two = 2
if len(options) < two:
return _format_error_response(
"INVALID_OPTIONS",
"Decision log requires at least 2 options",
)
async with httpx.AsyncClient() as client:
payload = {
"title": title,
"context": context,
"options": options,
"chosen": chosen,
"rationale": rationale,
"consequences": consequences or [],
"task_id": task_id,
"tags": tags or [],
}
resp = await client.post(
f"{_get_api_url()}/journals/me/decisions",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return _format_error_response(
"CREATE_FAILED",
"Failed to create decision log",
{"api_error": resp.text},
)
entry = resp.json()
return {
"status": "created",
"entry": entry,
"guidance": (
"Decision logged. If you need to revisit this decision later, "
"you'll have the context of why it was made."
),
}
# =========================================================================
# LEARNING
# =========================================================================
return await _handle_decision(data, agent_id)
@mcp.tool()
async def roboco_journal_learning(
title: str,
what_learned: str,
how_applied: str | None = None,
source: str | None = None,
task_id: str | None = None,
tags: list[str] | None = None,
) -> dict[str, Any]:
async def roboco_journal_learning(data: LearningInput) -> dict[str, Any]:
"""
Log something you learned.
Track learnings to:
- Build your knowledge base
- Help future you with similar problems
- Share knowledge with the team (if not private)
Args:
title: Learning title
what_learned: The actual learning/insight
how_applied: How you applied or plan to apply this
source: Where you learned this (docs, experiment, colleague, etc.)
task_id: Optional related task
tags: Optional list of tags
Returns:
Created learning entry
Track learnings to build your knowledge base and help future you.
"""
async with httpx.AsyncClient() as client:
payload = {
"title": title,
"what_learned": what_learned,
"how_applied": how_applied,
"source": source,
"task_id": task_id,
"tags": tags or [],
}
resp = await client.post(
f"{_get_api_url()}/journals/me/learnings",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return _format_error_response(
"CREATE_FAILED",
"Failed to create learning entry",
{"api_error": resp.text},
)
entry = resp.json()
return {
"status": "created",
"entry": entry,
"guidance": "Learning recorded. Use tags to make it searchable later.",
}
# =========================================================================
# STRUGGLE
# =========================================================================
return await _handle_learning(data, agent_id)
@mcp.tool()
async def roboco_journal_struggle(
title: str,
what_struggled: str,
attempted_solutions: list[str] | None = None,
resolution: str | None = None,
help_needed: str | None = None,
task_id: str | None = None,
tags: list[str] | None = None,
) -> dict[str, Any]:
async def roboco_journal_struggle(data: StruggleInput) -> dict[str, Any]:
"""
Log a struggle or challenge.
Recording struggles helps:
- Track problem-solving patterns
- Create documentation for others
- Get help if needed (help_needed field)
- Remember solutions for similar problems
Args:
title: Struggle title
what_struggled: What the challenge was
attempted_solutions: What you tried (even if it didn't work)
resolution: How it was resolved (if resolved)
help_needed: What help you need (if unresolved)
task_id: Optional related task
tags: Optional list of tags
Returns:
Created struggle entry
Recording struggles helps track problem-solving patterns and
create documentation for others.
"""
async with httpx.AsyncClient() as client:
payload = {
"title": title,
"what_struggled": what_struggled,
"attempted_solutions": attempted_solutions or [],
"resolution": resolution,
"help_needed": help_needed,
"task_id": task_id,
"tags": tags or [],
}
resp = await client.post(
f"{_get_api_url()}/journals/me/struggles",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [200, 201]:
return _format_error_response(
"CREATE_FAILED",
"Failed to create struggle entry",
{"api_error": resp.text},
)
entry = resp.json()
guidance = "Struggle recorded."
if help_needed and not resolution:
guidance += " Since you indicated help is needed, consider asking in your cell channel."
return {
"status": "created",
"entry": entry,
"guidance": guidance,
}
# =========================================================================
# SEARCH
# =========================================================================
return await _handle_struggle(data, agent_id)
@mcp.tool()
async def roboco_journal_search(
query: str,
top_k: int = 5,
) -> dict[str, Any]:
async def roboco_journal_search(query: str, top_k: int = 5) -> dict[str, Any]:
"""
Search your past journal entries.
Uses semantic search to find relevant entries based on meaning,
not just keywords. Great for:
- Finding past decisions on similar topics
- Recalling how you solved similar problems
- Getting context from previous work
Args:
query: What to search for
top_k: Maximum results to return (default 5)
Returns:
Matching journal entries
Uses semantic search to find relevant entries based on meaning.
"""
async with httpx.AsyncClient() as client:
payload = {
"query": query,
"top_k": min(top_k, 20), # Cap at 20
}
resp = await client.post(
f"{_get_api_url()}/journals/me/search",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"SEARCH_FAILED",
"Failed to search journal",
{"api_error": resp.text},
)
entries = resp.json()
if not entries:
return {
"entries": [],
"guidance": "No matching entries found. Try different keywords.",
}
return {
"entries": entries,
"count": len(entries),
"guidance": f"Found {len(entries)} relevant entries.",
}
# =========================================================================
# STATS
# =========================================================================
return await _handle_search(query, top_k, agent_id)
@mcp.tool()
async def roboco_journal_stats() -> dict[str, Any]:
@@ -499,56 +482,8 @@ def create_journal_mcp_server(agent_id: str) -> FastMCP:
Get statistics about your journal.
Returns counts by entry type, growth metrics, and other stats.
Useful for reflection and tracking your development.
Returns:
Journal statistics
"""
async with httpx.AsyncClient() as client:
# Get basic stats
stats_resp = await client.get(
f"{_get_api_url()}/journals/me/stats",
headers={"X-Agent-Id": agent_id},
)
# Get growth metrics
growth_resp = await client.get(
f"{_get_api_url()}/journals/me/growth",
headers={"X-Agent-Id": agent_id},
)
stats = (
stats_resp.json()
if stats_resp.status_code == status.HTTP_200_OK
else {}
)
growth = (
growth_resp.json()
if growth_resp.status_code == status.HTTP_200_OK
else {}
)
return {
"total_entries": stats.get("total_entries", 0),
"entries_by_type": stats.get("entries_by_type", {}),
"last_entry_at": stats.get("last_entry_at"),
"growth_metrics": {
"total_reflections": growth.get("total_reflections", 0),
"total_learnings": growth.get("total_learnings", 0),
"total_struggles": growth.get("total_struggles", 0),
"total_decisions": growth.get("total_decisions", 0),
"struggle_resolution_rate": growth.get("struggle_resolution_rate", 0),
"sentiment_trend": growth.get("sentiment_trend", "stable"),
},
"guidance": (
"These stats reflect your journal activity. "
"Regular journaling helps build context for future sessions."
),
}
# =========================================================================
# LIST RECENT
# =========================================================================
return await _handle_stats(agent_id)
@mcp.tool()
async def roboco_journal_recent(
@@ -559,43 +494,10 @@ def create_journal_mcp_server(agent_id: str) -> FastMCP:
"""
List recent journal entries.
Args:
entry_type:
Optional filter by type
(general, task_reflection, decision_log, learning, struggle)
task_id:
Optional filter by related task
limit:
Maximum entries to return
Returns:
Recent journal entries
Filter by entry_type (general, task_reflection, decision_log,
learning, struggle) or by task_id.
"""
async with httpx.AsyncClient() as client:
params: dict[str, Any] = {"limit": min(limit, 50)}
if entry_type:
params["entry_type"] = entry_type
if task_id:
params["task_id"] = task_id
resp = await client.get(
f"{_get_api_url()}/journals/me/entries",
params=params,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"LIST_FAILED",
"Failed to list entries",
)
entries = resp.json()
return {
"entries": entries,
"count": len(entries),
}
return await _handle_recent(entry_type, task_id, limit, agent_id)
return mcp
@@ -607,12 +509,11 @@ def create_journal_mcp_server(agent_id: str) -> FastMCP:
if __name__ == "__main__":
import sys
two = 2
if len(sys.argv) < two:
MIN_ARGS = 2
if len(sys.argv) < MIN_ARGS:
print("Usage: python journal_server.py <agent_id>")
sys.exit(1)
agent_id = sys.argv[1]
server = create_journal_mcp_server(agent_id)
agent_id_arg = sys.argv[1]
server = create_journal_mcp_server(agent_id_arg)
server.run()
roboco/mcp/message_server.py
+355 -367
View File
@@ -12,12 +12,14 @@ Tools:
- roboco_channel_history: Get channel message history
"""
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime, timedelta
from typing import Any
import httpx
from fastapi import status
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from roboco.agents_config import CHANNEL_ACCESS
from roboco.config import settings
@@ -27,6 +29,30 @@ from roboco.llm import ToonAdapter
_toon = ToonAdapter()
# =============================================================================
# INPUT MODELS
# =============================================================================
class SendMessageInput(BaseModel):
"""Input for sending a message."""
channel_slug: str = Field(..., description="Channel slug (e.g., 'backend-cell')")
content: str = Field(..., description="Message content")
message_type: str = Field(
default="dialogue",
description="Type: reasoning, dialogue, decision, action, blocker, technical",
)
task_id: str | None = Field(default=None, description="Optional related task ID")
reply_to: str | None = Field(default=None, description="Message ID to reply to")
mentions: list[str] = Field(default_factory=list, description="Agents to mention")
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _check_channel_access(agent_id: str, channel_slug: str, action: str) -> bool:
"""Check if agent has access to channel for the given action."""
channel = CHANNEL_ACCESS.get(channel_slug, {})
@@ -41,11 +67,6 @@ def _check_channel_access(agent_id: str, channel_slug: str, action: str) -> bool
return bool(action == "read" and agent_id in channel.get("silent", []))
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _get_api_url() -> str:
"""Get the RoboCo API base URL."""
return f"http://{settings.host}:{settings.port}/api/v1"
@@ -66,6 +87,304 @@ def _format_error_response(
}
def _validate_message_send(
agent_id: str,
channel_slug: str,
content: str,
message_type: str,
) -> dict[str, Any] | None:
"""Validate message send parameters. Returns error dict or None if valid."""
valid_types = [
"reasoning",
"dialogue",
"decision",
"action",
"blocker",
"technical",
]
if message_type not in valid_types:
return _format_error_response(
"INVALID_TYPE",
f"Invalid message type '{message_type}'. Must be one of: {valid_types}",
)
if not _check_channel_access(agent_id, channel_slug, "write"):
writable = [
ch for ch in CHANNEL_ACCESS if _check_channel_access(agent_id, ch, "write")
]
return _format_error_response(
"ACCESS_DENIED",
f"You don't have write access to #{channel_slug}",
{"your_writable_channels": writable},
)
if agent_id in CHANNEL_ACCESS.get(channel_slug, {}).get("silent", []):
return _format_error_response(
"SILENT_OBSERVER",
"You are a silent observer on this channel and cannot post messages.",
)
if not content or not content.strip():
return _format_error_response(
"EMPTY_CONTENT", "Message content cannot be empty."
)
return None
async def _get_or_create_session(
client: httpx.AsyncClient,
channel_id: str,
) -> str | dict[str, Any]:
"""Get or create session for channel. Returns session_id or error dict."""
session_resp = await client.get(f"{_get_api_url()}/channels/{channel_id}/session")
if session_resp.status_code == status.HTTP_200_OK:
return str(session_resp.json()["id"])
create_resp = await client.post(
f"{_get_api_url()}/sessions",
json={"channel_id": channel_id},
)
if create_resp.status_code in [status.HTTP_200_OK, status.HTTP_201_CREATED]:
return str(create_resp.json()["id"])
return _format_error_response("SESSION_ERROR", "Failed to get or create session")
# =============================================================================
# TOOL IMPLEMENTATIONS
# =============================================================================
async def _handle_channel_list(agent_id: str) -> dict[str, Any]:
"""Handle channel listing."""
readable = []
writable = []
for channel_slug in CHANNEL_ACCESS:
if _check_channel_access(agent_id, channel_slug, "read"):
readable.append(channel_slug)
if _check_channel_access(agent_id, channel_slug, "write"):
writable.append(channel_slug)
guidance = (
f"You can read from {len(readable)} channel(s) and "
f"write to {len(writable)} channel(s). "
"Use roboco_message_send to post messages. "
"Use roboco_channel_history to read recent messages."
)
return {
"readable_channels": readable,
"writable_channels": writable,
"guidance": guidance,
}
async def _handle_channel_history(
agent_id: str,
channel_slug: str,
limit: int,
hours_back: int,
) -> dict[str, Any]:
"""Handle channel history retrieval."""
if not _check_channel_access(agent_id, channel_slug, "read"):
return _format_error_response(
"ACCESS_DENIED", f"You don't have read access to #{channel_slug}"
)
limit = min(limit, 100)
since = datetime.now(UTC) - timedelta(hours=hours_back)
async with httpx.AsyncClient() as client:
channels_resp = await client.get(
f"{_get_api_url()}/channels",
params={"slug": channel_slug},
)
if channels_resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch channels")
channels = channels_resp.json()
if not channels:
return _format_error_response(
"NOT_FOUND", f"Channel #{channel_slug} not found"
)
channel_id = channels[0]["id"]
messages_resp = await client.get(
f"{_get_api_url()}/channels/{channel_id}/messages",
params={"after": since.isoformat(), "limit": limit},
)
if messages_resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch messages")
messages = messages_resp.json()
return {
"channel": channel_slug,
"messages": messages.get("items", []),
"total": messages.get("total", 0),
"has_more": messages.get("has_more", False),
"since": since.isoformat(),
}
async def _handle_message_send(
agent_id: str,
data: SendMessageInput,
) -> dict[str, Any]:
"""Handle message sending."""
if validation_error := _validate_message_send(
agent_id, data.channel_slug, data.content, data.message_type
):
return validation_error
async with httpx.AsyncClient() as client:
channels_resp = await client.get(
f"{_get_api_url()}/channels",
params={"slug": data.channel_slug},
)
if channels_resp.status_code != status.HTTP_200_OK or not channels_resp.json():
return _format_error_response(
"NOT_FOUND", f"Channel #{data.channel_slug} not found"
)
channel = channels_resp.json()[0]
channel_id = channel["id"]
session_result = await _get_or_create_session(client, channel_id)
if isinstance(session_result, dict):
return session_result
session_id = session_result
message_data = {
"session_id": session_id,
"type": data.message_type,
"content": data.content,
"is_reply": data.reply_to is not None,
"reply_to": data.reply_to,
"mentions": data.mentions,
"task_id": data.task_id,
}
send_resp = await client.post(
f"{_get_api_url()}/messages",
json=message_data,
headers={"X-Agent-Id": agent_id},
)
if send_resp.status_code not in [status.HTTP_200_OK, status.HTTP_201_CREATED]:
return _format_error_response(
"SEND_FAILED", "Failed to send message", {"api_error": send_resp.text}
)
return {
"status": "sent",
"message": send_resp.json(),
"channel": data.channel_slug,
"guidance": "Message sent successfully.",
}
async def _handle_message_get(message_id: str) -> dict[str, Any]:
"""Handle message retrieval."""
async with httpx.AsyncClient() as client:
resp = await client.get(f"{_get_api_url()}/messages/{message_id}")
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response(
"NOT_FOUND", f"Message {message_id} not found"
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch message")
return {"message": resp.json()}
class AskQuestionInput(BaseModel):
"""Input for asking a question."""
channel_slug: str
question: str
context: str | None = None
task_id: str | None = None
class ReportBlockerInput(BaseModel):
"""Input for reporting a blocker."""
channel_slug: str
blocker_description: str
what_needed: str
task_id: str | None = None
async def _handle_ask_question(
data: AskQuestionInput,
send_fn: Callable[[SendMessageInput], Awaitable[dict[str, Any]]],
) -> dict[str, Any]:
"""Handle asking a question."""
content = f"**Question**: {data.question}"
if data.context:
content = f"{data.context}\n\n{content}"
msg_data = SendMessageInput(
channel_slug=data.channel_slug,
content=content,
message_type="dialogue",
task_id=data.task_id,
)
result = await send_fn(msg_data)
if "error" in result:
return result
result["guidance"] = (
"Question posted. You should now:\n"
"1. Wait for an answer before proceeding with related work\n"
"2. Check roboco_channel_history periodically for responses\n"
"3. If urgent, consider mentioning the PM"
)
return result
async def _handle_report_blocker(
data: ReportBlockerInput,
send_fn: Callable[[SendMessageInput], Awaitable[dict[str, Any]]],
) -> dict[str, Any]:
"""Handle reporting a blocker."""
content = (
f"**BLOCKER**\n\n"
f"**Issue**: {data.blocker_description}\n\n"
f"**Needed to unblock**: {data.what_needed}"
)
msg_data = SendMessageInput(
channel_slug=data.channel_slug,
content=content,
message_type="blocker",
task_id=data.task_id,
)
result = await send_fn(msg_data)
if "error" in result:
return result
result["guidance"] = (
"Blocker reported. The PM will be notified.\n"
"You should:\n"
"1. Wait for resolution, or\n"
"2. Switch to another task (call roboco_task_scan)"
)
return result
# =============================================================================
# MCP SERVER FACTORY
# =============================================================================
@@ -75,8 +394,6 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
"""
Create a Message MCP server for a specific agent.
The agent_id is embedded in the server to enforce access rules.
Args:
agent_id: The agent identifier (e.g., "be-dev-1")
@@ -85,40 +402,10 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
"""
mcp = FastMCP(f"roboco-message-{agent_id}", json_response=True)
# =========================================================================
# CHANNEL LISTING
# =========================================================================
@mcp.tool()
async def roboco_channel_list() -> dict[str, Any]:
"""
List channels you have access to.
Returns:
Dict with readable and writable channels
"""
readable = []
writable = []
for channel_slug, _access in CHANNEL_ACCESS.items():
if _check_channel_access(agent_id, channel_slug, "read"):
readable.append(channel_slug)
if _check_channel_access(agent_id, channel_slug, "write"):
writable.append(channel_slug)
return {
"readable_channels": readable,
"writable_channels": writable,
"guidance": (
f"You can read from {len(readable)} channel(s) and write to {len(writable)} channel(s). "
"Use roboco_message_send to post messages. "
"Use roboco_channel_history to read recent messages."
),
}
# =========================================================================
# CHANNEL HISTORY
# =========================================================================
"""List channels you have access to."""
return await _handle_channel_list(agent_id)
@mcp.tool()
async def roboco_channel_history(
@@ -129,270 +416,23 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
"""
Get recent message history from a channel.
ENFORCEMENT:
- You must have read access to the channel
Args:
channel_slug: The channel slug (e.g., "backend-cell")
limit: Maximum messages to return (default 50, max 100)
hours_back: How many hours back to look (default 24)
Returns:
List of messages with metadata
You must have read access to the channel.
"""
# Check read access
if not _check_channel_access(agent_id, channel_slug, "read"):
return _format_error_response(
"ACCESS_DENIED",
f"You don't have read access to #{channel_slug}",
)
limit = min(limit, 100)
since = datetime.now(UTC) - timedelta(hours=hours_back)
async with httpx.AsyncClient() as client:
# Get channel ID from slug
channels_resp = await client.get(
f"{_get_api_url()}/channels",
params={"slug": channel_slug},
)
if channels_resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch channels")
channels = channels_resp.json()
if not channels:
return _format_error_response(
"NOT_FOUND", f"Channel #{channel_slug} not found"
)
channel_id = channels[0]["id"]
# Get messages
messages_resp = await client.get(
f"{_get_api_url()}/channels/{channel_id}/messages",
params={
"after": since.isoformat(),
"limit": limit,
},
)
if messages_resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch messages")
messages = messages_resp.json()
return {
"channel": channel_slug,
"messages": messages.get("items", []),
"total": messages.get("total", 0),
"has_more": messages.get("has_more", False),
"since": since.isoformat(),
}
# =========================================================================
# SEND MESSAGE
# =========================================================================
def _validate_message_send(
channel_slug: str,
content: str,
message_type: str,
) -> dict[str, Any] | None:
"""Validate message send parameters. Returns error dict or None if valid."""
valid_types = [
"reasoning",
"dialogue",
"decision",
"action",
"blocker",
"technical",
]
if message_type not in valid_types:
return _format_error_response(
"INVALID_TYPE",
f"Invalid message type '{message_type}'. Must be one of: {valid_types}",
)
if not _check_channel_access(agent_id, channel_slug, "write"):
return _format_error_response(
"ACCESS_DENIED",
f"You don't have write access to #{channel_slug}",
{
"your_writable_channels": [
ch
for ch in CHANNEL_ACCESS
if _check_channel_access(agent_id, ch, "write")
]
},
)
if agent_id in CHANNEL_ACCESS.get(channel_slug, {}).get("silent", []):
return _format_error_response(
"SILENT_OBSERVER",
"You are a silent observer on this channel and cannot post messages.",
)
if not content or not content.strip():
return _format_error_response(
"EMPTY_CONTENT",
"Message content cannot be empty.",
)
return None
async def _get_or_create_session(
client: httpx.AsyncClient,
channel_id: str,
) -> str | dict[str, Any]:
"""Get or create session for channel. Returns session_id or error dict."""
session_resp = await client.get(
f"{_get_api_url()}/channels/{channel_id}/session",
)
if session_resp.status_code == status.HTTP_200_OK:
session_id: str = session_resp.json()["id"]
return session_id
create_resp = await client.post(
f"{_get_api_url()}/sessions",
json={"channel_id": channel_id},
)
if create_resp.status_code in [status.HTTP_200_OK, status.HTTP_201_CREATED]:
created_id: str = create_resp.json()["id"]
return created_id
return _format_error_response(
"SESSION_ERROR", "Failed to get or create session"
)
return await _handle_channel_history(agent_id, channel_slug, limit, hours_back)
@mcp.tool()
async def roboco_message_send(
channel_slug: str,
content: str,
message_type: str = "dialogue",
task_id: str | None = None,
reply_to: str | None = None,
mentions: list[str] | None = None,
) -> dict[str, Any]:
async def roboco_message_send(data: SendMessageInput) -> dict[str, Any]:
"""
Send a message to a channel.
ENFORCEMENT:
- You must have write access to the channel
- Message type must be valid
- Content is required
Args:
channel_slug: The channel slug (e.g., "backend-cell")
content: Message content
message_type: Type of message (reasoning, dialogue, decision, action, blocker, technical)
task_id: Optional task ID this message relates to
reply_to: Optional message ID to reply to
mentions: Optional list of agent IDs to mention (adds @agent-id)
Returns:
Sent message with confirmation
You must have write access to the channel.
"""
# Validate inputs
if validation_error := _validate_message_send(
channel_slug, content, message_type
):
return validation_error
async with httpx.AsyncClient() as client:
# Get channel
channels_resp = await client.get(
f"{_get_api_url()}/channels",
params={"slug": channel_slug},
)
if (
channels_resp.status_code != status.HTTP_200_OK
or not channels_resp.json()
):
return _format_error_response(
"NOT_FOUND", f"Channel #{channel_slug} not found"
)
channel = channels_resp.json()[0]
channel_id = channel["id"]
# Get or create session
session_result = await _get_or_create_session(client, channel_id)
if isinstance(session_result, dict):
return session_result # Error response
session_id = session_result
# Build and send message
message_data = {
"session_id": session_id,
"type": message_type,
"content": content,
"is_reply": reply_to is not None,
"reply_to": reply_to,
"mentions": mentions or [],
"task_id": task_id,
}
send_resp = await client.post(
f"{_get_api_url()}/messages",
json=message_data,
headers={"X-Agent-Id": agent_id},
)
if send_resp.status_code not in [
status.HTTP_200_OK,
status.HTTP_201_CREATED,
]:
return _format_error_response(
"SEND_FAILED",
"Failed to send message",
{"api_error": send_resp.text},
)
return {
"status": "sent",
"message": send_resp.json(),
"channel": channel_slug,
"guidance": "Message sent successfully.",
}
# =========================================================================
# GET MESSAGE
# =========================================================================
return await _handle_message_send(agent_id, data)
@mcp.tool()
async def roboco_message_get(message_id: str) -> dict[str, Any]:
"""
Get a specific message by ID.
Args:
message_id: The message UUID
Returns:
Message details
"""
async with httpx.AsyncClient() as client:
resp = await client.get(f"{_get_api_url()}/messages/{message_id}")
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response(
"NOT_FOUND", f"Message {message_id} not found"
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch message")
message = resp.json()
return {
"message": message,
}
# =========================================================================
# ASK QUESTION (convenience wrapper)
# =========================================================================
"""Get a specific message by ID."""
return await _handle_message_get(message_id)
@mcp.tool()
async def roboco_ask_question(
@@ -402,49 +442,21 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
task_id: str | None = None,
) -> dict[str, Any]:
"""
Ask a question in a channel (convenience wrapper).
Ask a question in a channel.
This is a common pattern - asking for clarification. The message
is automatically formatted as a question.
IMPORTANT: After asking, you should wait for an answer before
proceeding with work that depends on this question.
Args:
channel_slug: The channel to ask in
question: The question to ask
context: Optional context for the question
task_id: Optional task this relates to
Returns:
Sent question message
After asking, wait for an answer before proceeding.
"""
content = f"**Question**: {question}"
if context:
content = f"{context}\n\n{content}"
result: dict[str, Any] = await roboco_message_send(
async def send_fn(d: SendMessageInput) -> dict[str, Any]:
return await _handle_message_send(agent_id, d)
data = AskQuestionInput(
channel_slug=channel_slug,
content=content,
message_type="dialogue",
question=question,
context=context,
task_id=task_id,
)
if "error" in result:
return result
result["guidance"] = (
"Question posted. You should now:\n"
"1. Wait for an answer before proceeding with related work\n"
"2. Check roboco_channel_history periodically for responses\n"
"3. If urgent, consider mentioning the PM"
)
return result
# =========================================================================
# REPORT BLOCKER (convenience wrapper)
# =========================================================================
return await _handle_ask_question(data, send_fn)
@mcp.tool()
async def roboco_report_blocker(
@@ -454,44 +466,21 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
task_id: str | None = None,
) -> dict[str, Any]:
"""
Report a blocker in a channel (convenience wrapper).
Report a blocker in a channel.
This automatically formats the message as a blocker report
and notifies the PM.
Args:
channel_slug: The channel to report in
blocker_description: What is blocking you
what_needed: What is needed to unblock
task_id: Optional task this relates to
Returns:
Sent blocker message
The PM will be notified automatically.
"""
content = (
f"**BLOCKER**\n\n"
f"**Issue**: {blocker_description}\n\n"
f"**Needed to unblock**: {what_needed}"
)
result: dict[str, Any] = await roboco_message_send(
async def send_fn(d: SendMessageInput) -> dict[str, Any]:
return await _handle_message_send(agent_id, d)
data = ReportBlockerInput(
channel_slug=channel_slug,
content=content,
message_type="blocker",
blocker_description=blocker_description,
what_needed=what_needed,
task_id=task_id,
)
if "error" in result:
return result
result["guidance"] = (
"Blocker reported. The PM will be notified.\n"
"You should:\n"
"1. Wait for resolution, or\n"
"2. Switch to another task (call roboco_task_scan)"
)
return result
return await _handle_report_blocker(data, send_fn)
return mcp
@@ -503,12 +492,11 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
if __name__ == "__main__":
import sys
two = 2
if len(sys.argv) < two:
MIN_ARGS = 2
if len(sys.argv) < MIN_ARGS:
print("Usage: python message_server.py <agent_id>")
sys.exit(1)
agent_id = sys.argv[1]
server = create_message_mcp_server(agent_id)
agent_id_arg = sys.argv[1]
server = create_message_mcp_server(agent_id_arg)
server.run()
roboco/mcp/notify_server.py
+240 -315
View File
@@ -16,6 +16,7 @@ from typing import Any
import httpx
from fastapi import status
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from roboco.agents_config import (
NOTIFICATION_PERMISSIONS,
@@ -24,14 +25,32 @@ from roboco.agents_config import (
)
from roboco.config import settings
# =============================================================================
# INPUT MODELS
# =============================================================================
class SendNotificationInput(BaseModel):
"""Input for sending a notification."""
recipients: list[str] = Field(..., description="Agent IDs to notify")
subject: str = Field(..., description="Notification subject")
body: str = Field(..., description="Notification body")
notification_type: str = Field(
default="info", description="Type: info, alert, task, escalation, approval"
)
priority: str = Field(default="normal", description="low, normal, high, urgent")
requires_ack: bool = Field(default=True, description="Require acknowledgment")
related_task_id: str | None = Field(default=None, description="Related task")
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str]:
"""
Check if sender can send notification to recipient.
Returns:
Tuple of (can_send, reason)
"""
"""Check if sender can send notification to recipient."""
role = get_agent_role(sender_id)
permissions = NOTIFICATION_PERMISSIONS.get(role, {"can_send": False})
@@ -60,11 +79,6 @@ def _can_send_notification(sender_id: str, recipient_id: str) -> tuple[bool, str
return False, f"You cannot send notifications to {recipient_id}"
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _get_api_url() -> str:
"""Get the RoboCo API base URL."""
return f"http://{settings.host}:{settings.port}/api/v1"
@@ -85,6 +99,198 @@ def _format_error_response(
}
# =============================================================================
# TOOL IMPLEMENTATIONS
# =============================================================================
async def _handle_list(
agent_id: str,
unread_only: bool,
pending_ack_only: bool,
limit: int,
) -> dict[str, Any]:
"""Handle notification listing."""
async with httpx.AsyncClient() as client:
params: dict[str, str | int] = {
"unread_only": str(unread_only).lower(),
"pending_ack_only": str(pending_ack_only).lower(),
"limit": limit,
}
resp = await client.get(
f"{_get_api_url()}/notifications",
params=params,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch notifications")
data = resp.json()
unread = data.get("unread_count", 0)
pending_ack = data.get("pending_ack_count", 0)
guidance_parts = []
if pending_ack > 0:
guidance_parts.append(
f"You have {pending_ack} notification(s) requiring acknowledgment. "
"Use roboco_notify_ack to acknowledge them."
)
if unread > 0:
guidance_parts.append(f"You have {unread} unread notification(s).")
if not guidance_parts:
guidance_parts.append("No new notifications.")
return {
"notifications": data.get("items", []),
"total": data.get("total", 0),
"unread_count": unread,
"pending_ack_count": pending_ack,
"guidance": " ".join(guidance_parts),
}
async def _handle_get(agent_id: str, notification_id: str) -> dict[str, Any]:
"""Handle getting a specific notification."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{_get_api_url()}/notifications/{notification_id}",
headers={"X-Agent-Id": agent_id},
)
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response("NOT_FOUND", "Notification not found")
if resp.status_code == status.HTTP_403_FORBIDDEN:
return _format_error_response(
"NOT_RECIPIENT", "You are not a recipient of this notification"
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response("API_ERROR", "Failed to fetch notification")
notification = resp.json()
guidance = ""
if notification.get("requires_ack") and not notification.get("is_acknowledged"):
guidance = (
"This notification requires acknowledgment. "
"Use roboco_notify_ack to acknowledge."
)
return {"notification": notification, "guidance": guidance}
async def _handle_ack(agent_id: str, notification_id: str) -> dict[str, Any]:
"""Handle acknowledging a notification."""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{_get_api_url()}/notifications/{notification_id}/ack",
headers={"X-Agent-Id": agent_id},
)
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response("NOT_FOUND", "Notification not found")
if resp.status_code == status.HTTP_403_FORBIDDEN:
return _format_error_response(
"NOT_RECIPIENT", "You are not a recipient of this notification"
)
if resp.status_code == status.HTTP_400_BAD_REQUEST:
return _format_error_response(
"NO_ACK_REQUIRED", "This notification does not require acknowledgment"
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"API_ERROR", "Failed to acknowledge notification"
)
notification = resp.json()
return {
"status": "acknowledged",
"notification": notification,
"guidance": "Notification acknowledged. The sender will be informed.",
}
async def _handle_send(agent_id: str, data: SendNotificationInput) -> dict[str, Any]:
"""Handle sending a notification."""
role = get_agent_role(agent_id)
permissions = NOTIFICATION_PERMISSIONS.get(role, {"can_send": False})
if not permissions.get("can_send", False):
return _format_error_response(
"NOT_AUTHORIZED",
f"Agents with role '{role}' cannot send notifications. "
"Only PMs, Board members, and Auditor can send notifications.",
{"your_role": role},
)
denied_recipients = []
for recipient in data.recipients:
can_send, reason = _can_send_notification(agent_id, recipient)
if not can_send:
denied_recipients.append({"recipient": recipient, "reason": reason})
if denied_recipients:
return _format_error_response(
"RECIPIENT_DENIED",
"Cannot send to one or more recipients",
{"denied": denied_recipients},
)
valid_types = ["info", "alert", "task", "escalation", "approval"]
if data.notification_type not in valid_types:
return _format_error_response(
"INVALID_TYPE", f"Invalid notification type. Must be one of: {valid_types}"
)
valid_priorities = ["low", "normal", "high", "urgent"]
if data.priority not in valid_priorities:
return _format_error_response(
"INVALID_PRIORITY", f"Invalid priority. Must be one of: {valid_priorities}"
)
async with httpx.AsyncClient() as client:
payload = {
"type": data.notification_type,
"priority": data.priority,
"to_agents": data.recipients,
"subject": data.subject,
"body": data.body,
"requires_ack": data.requires_ack,
"related_task_id": data.related_task_id,
}
resp = await client.post(
f"{_get_api_url()}/notifications",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [status.HTTP_200_OK, status.HTTP_201_CREATED]:
return _format_error_response(
"SEND_FAILED", "Failed to send notification", {"api_error": resp.text}
)
notification = resp.json()
ack_note = "Recipients must acknowledge." if data.requires_ack else ""
count = len(data.recipients)
return {
"status": "sent",
"notification": notification,
"recipients_count": count,
"guidance": f"Notification sent to {count} recipient(s). {ack_note}".strip(),
}
# =============================================================================
# MCP SERVER FACTORY
# =============================================================================
@@ -94,8 +300,6 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
"""
Create a Notify MCP server for a specific agent.
The agent_id is embedded in the server to enforce permissions.
Args:
agent_id: The agent identifier (e.g., "be-pm")
@@ -104,288 +308,34 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
"""
mcp = FastMCP(f"roboco-notify-{agent_id}", json_response=True)
# =========================================================================
# LIST NOTIFICATIONS
# =========================================================================
@mcp.tool()
async def roboco_notify_list(
unread_only: bool = False,
pending_ack_only: bool = False,
limit: int = 50,
) -> dict[str, Any]:
"""
List your notifications.
Args:
unread_only: Only show unread notifications
pending_ack_only: Only show notifications pending acknowledgment
limit: Maximum notifications to return
Returns:
List of notifications with counts
"""
async with httpx.AsyncClient() as client:
params: dict[str, str | int] = {
"unread_only": str(unread_only).lower(),
"pending_ack_only": str(pending_ack_only).lower(),
"limit": limit,
}
resp = await client.get(
f"{_get_api_url()}/notifications",
params=params,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"API_ERROR", "Failed to fetch notifications"
)
data = resp.json()
# Add guidance based on counts
unread = data.get("unread_count", 0)
pending_ack = data.get("pending_ack_count", 0)
guidance_parts = []
if pending_ack > 0:
guidance_parts.append(
f"You have {pending_ack} notification(s) requiring acknowledgment. "
"Use roboco_notify_ack to acknowledge them."
)
if unread > 0:
guidance_parts.append(f"You have {unread} unread notification(s).")
if not guidance_parts:
guidance_parts.append("No new notifications.")
return {
"notifications": data.get("items", []),
"total": data.get("total", 0),
"unread_count": unread,
"pending_ack_count": pending_ack,
"guidance": " ".join(guidance_parts),
}
# =========================================================================
# GET NOTIFICATION
# =========================================================================
"""List your notifications."""
return await _handle_list(agent_id, unread_only, pending_ack_only, limit)
@mcp.tool()
async def roboco_notify_get(notification_id: str) -> dict[str, Any]:
"""
Get a specific notification.
This also marks the notification as read.
Args:
notification_id: The notification UUID
Returns:
Notification details
"""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{_get_api_url()}/notifications/{notification_id}",
headers={"X-Agent-Id": agent_id},
)
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response("NOT_FOUND", "Notification not found")
if resp.status_code == status.HTTP_403_FORBIDDEN:
return _format_error_response(
"NOT_RECIPIENT",
"You are not a recipient of this notification",
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"API_ERROR", "Failed to fetch notification"
)
notification = resp.json()
guidance = ""
if notification.get("requires_ack") and not notification.get("is_acknowledged"):
guidance = (
"This notification requires acknowledgment. "
"Use roboco_notify_ack to acknowledge."
)
return {
"notification": notification,
"guidance": guidance,
}
# =========================================================================
# ACKNOWLEDGE NOTIFICATION
# =========================================================================
"""Get a specific notification. Also marks it as read."""
return await _handle_get(agent_id, notification_id)
@mcp.tool()
async def roboco_notify_ack(notification_id: str) -> dict[str, Any]:
"""
Acknowledge a notification.
Some notifications require acknowledgment to confirm receipt
and understanding.
Args:
notification_id: The notification UUID
Returns:
Updated notification
"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{_get_api_url()}/notifications/{notification_id}/ack",
headers={"X-Agent-Id": agent_id},
)
if resp.status_code == status.HTTP_404_NOT_FOUND:
return _format_error_response("NOT_FOUND", "Notification not found")
if resp.status_code == status.HTTP_403_FORBIDDEN:
return _format_error_response(
"NOT_RECIPIENT",
"You are not a recipient of this notification",
)
if resp.status_code == status.HTTP_400_BAD_REQUEST:
return _format_error_response(
"NO_ACK_REQUIRED",
"This notification does not require acknowledgment",
)
if resp.status_code != status.HTTP_200_OK:
return _format_error_response(
"API_ERROR", "Failed to acknowledge notification"
)
notification = resp.json()
return {
"status": "acknowledged",
"notification": notification,
"guidance": "Notification acknowledged. The sender will be informed.",
}
# =========================================================================
# SEND NOTIFICATION (PM/Board/Auditor only)
# =========================================================================
"""Acknowledge a notification."""
return await _handle_ack(agent_id, notification_id)
@mcp.tool()
async def roboco_notify_send(
recipients: list[str],
subject: str,
body: str,
notification_type: str = "info",
priority: str = "normal",
requires_ack: bool = True,
related_task_id: str | None = None,
) -> dict[str, Any]:
async def roboco_notify_send(data: SendNotificationInput) -> dict[str, Any]:
"""
Send a notification to one or more agents.
ENFORCEMENT:
- Only PMs, Board members, and Auditor can send notifications
- Cell PMs can only notify their own cell
- Developers, QA, and Documenters CANNOT send notifications
Args:
recipients: List of agent IDs to notify
subject: Notification subject
body: Notification body
notification_type: Type (info, alert, task, escalation, approval)
priority: Priority (low, normal, high, urgent)
requires_ack: Whether recipients must acknowledge
related_task_id: Optional related task
Returns:
Sent notification or error
Only PMs, Board members, and Auditor can send notifications.
Cell PMs can only notify their own cell.
"""
# Check sender permissions
role = get_agent_role(agent_id)
permissions = NOTIFICATION_PERMISSIONS.get(role, {"can_send": False})
if not permissions.get("can_send", False):
return _format_error_response(
"NOT_AUTHORIZED",
f"Agents with role '{role}' cannot send notifications. "
"Only PMs, Board members, and Auditor can send notifications.",
{"your_role": role},
)
# Check each recipient
denied_recipients = []
for recipient in recipients:
can_send, reason = _can_send_notification(agent_id, recipient)
if not can_send:
denied_recipients.append({"recipient": recipient, "reason": reason})
if denied_recipients:
return _format_error_response(
"RECIPIENT_DENIED",
"Cannot send to one or more recipients",
{"denied": denied_recipients},
)
# Validate notification type
valid_types = ["info", "alert", "task", "escalation", "approval"]
if notification_type not in valid_types:
return _format_error_response(
"INVALID_TYPE",
f"Invalid notification type. Must be one of: {valid_types}",
)
# Validate priority
valid_priorities = ["low", "normal", "high", "urgent"]
if priority not in valid_priorities:
return _format_error_response(
"INVALID_PRIORITY",
f"Invalid priority. Must be one of: {valid_priorities}",
)
async with httpx.AsyncClient() as client:
payload = {
"type": notification_type,
"priority": priority,
"to_agents": recipients,
"subject": subject,
"body": body,
"requires_ack": requires_ack,
"related_task_id": related_task_id,
}
resp = await client.post(
f"{_get_api_url()}/notifications",
json=payload,
headers={"X-Agent-Id": agent_id},
)
if resp.status_code not in [status.HTTP_200_OK, status.HTTP_201_CREATED]:
return _format_error_response(
"SEND_FAILED",
"Failed to send notification",
{"api_error": resp.text},
)
notification = resp.json()
ack_note = "Recipients must acknowledge." if requires_ack else ""
return {
"status": "sent",
"notification": notification,
"recipients_count": len(recipients),
"guidance": f"Notification sent to {len(recipients)} recipient(s). {ack_note}",
}
# =========================================================================
# CONVENIENCE: ESCALATE (PM only)
# =========================================================================
return await _handle_send(agent_id, data)
@mcp.tool()
async def roboco_escalate(
@@ -395,27 +345,17 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
task_id: str | None = None,
) -> dict[str, Any]:
"""
Escalate an issue to a higher level (PM convenience wrapper).
Escalate an issue to a higher level (PM only).
This sends a high-priority notification requiring acknowledgment.
Args:
escalate_to: Agent ID to escalate to (e.g., "main-pm")
subject: Escalation subject
description: Detailed description of the issue
task_id: Optional related task
Returns:
Sent escalation notification
Sends a high-priority notification requiring acknowledgment.
"""
role = get_agent_role(agent_id)
if role not in ["cell_pm", "main_pm"]:
return _format_error_response(
"NOT_PM",
"Only PMs can use the escalate function",
"NOT_PM", "Only PMs can use the escalate function"
)
result: dict[str, Any] = await roboco_notify_send(
input_data = SendNotificationInput(
recipients=[escalate_to],
subject=f"[ESCALATION] {subject}",
body=description,
@@ -424,11 +364,7 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
requires_ack=True,
related_task_id=task_id,
)
return result
# =========================================================================
# CONVENIENCE: REQUEST APPROVAL (PM/Board only)
# =========================================================================
return await _handle_send(agent_id, input_data)
@mcp.tool()
async def roboco_request_approval(
@@ -438,25 +374,15 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
task_id: str | None = None,
) -> dict[str, Any]:
"""
Request approval from someone (PM/Board convenience wrapper).
Args:
approver: Agent ID to request approval from
subject: Approval subject
what_needs_approval: Description of what needs approval
task_id: Optional related task
Returns:
Sent approval request notification
Request approval from someone (PM/Board only).
"""
role = get_agent_role(agent_id)
if role not in ["cell_pm", "main_pm", "product_owner", "head_marketing"]:
return _format_error_response(
"NOT_AUTHORIZED",
"Only PMs and Board can request approvals",
"NOT_AUTHORIZED", "Only PMs and Board can request approvals"
)
result: dict[str, Any] = await roboco_notify_send(
input_data = SendNotificationInput(
recipients=[approver],
subject=f"[APPROVAL NEEDED] {subject}",
body=what_needs_approval,
@@ -465,7 +391,7 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
requires_ack=True,
related_task_id=task_id,
)
return result
return await _handle_send(agent_id, input_data)
return mcp
@@ -477,12 +403,11 @@ def create_notify_mcp_server(agent_id: str) -> FastMCP:
if __name__ == "__main__":
import sys
two = 2
if len(sys.argv) < two:
MIN_ARGS = 2
if len(sys.argv) < MIN_ARGS:
print("Usage: python notify_server.py <agent_id>")
sys.exit(1)
agent_id = sys.argv[1]
server = create_notify_mcp_server(agent_id)
agent_id_arg = sys.argv[1]
server = create_notify_mcp_server(agent_id_arg)
server.run()
roboco/mcp/task_server.py
+813 -725
View File
File diff suppressed because it is too large Load Diff