1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
#!/usr/bin/env python3
"""Sénia MCP server: shared registration point for all senia tools.
Tools are implemented in their respective submodules and re-exported via
the package __init__. This file is intentionally thin: it only handles
registration and startup.
"""
import functools
from db import run_migrations
from mcp.server.fastmcp import FastMCP
import diary
import tasks
mcp = FastMCP("senia")
@mcp.tool()
def diary_log(content: str) -> dict:
"""Record a brief observation, decision, or moment to your diary.
Use generously throughout a conversation: things the user mentioned, decisions
made, patterns you noticed, your own reactions. Keep entries short (one or two
sentences): granular is fine, you can write many. These become the raw material
for daily reflection and longer-term memory.
Args:
content: The observation, in your own words. Be specific; "user mentioned X"
is more useful than "user is busy".
Returns:
The created entry.
"""
return diary.log(content)
@mcp.tool()
def diary_today() -> list[dict]:
"""Return all diary entries logged so far today (since 00:00 local time).
Use this when starting a new conversation to recall what you've already
noted today, so you don't rediscover the same conclusions.
"""
return diary.today()
@mcp.tool()
def diary_recent(hours: int = 24) -> list[dict]:
"""Return diary entries from the last N hours. Default 24."""
return diary.recent(hours)
@mcp.tool()
def diary_range(start_ts: str, end_ts: str) -> list[dict]:
"""Return diary entries between two ISO 8601 timestamps (inclusive start,
exclusive end). Useful for inspecting specific past periods."""
return diary.range_query(start_ts, end_ts)
@mcp.tool()
def tasks_add(owner: str, content: str, due_at: str | None = None,
notes: str | None = None) -> dict:
"""Create a live task.
owner determines handling:
- "agent": something you can work on autonomously; pick it up, do it, then complete it.
- "user": something only the user can do (appointments, errands, calls). You cannot
finish these yourself; your job is to gently remind the user until they confirm done.
Args:
owner: "agent" or "user".
content: What the task is, in natural language.
due_at: Optional ISO 8601 deadline. Drives reminder urgency for user tasks.
notes: Optional context.
"""
return tasks.add_task(owner, content, due_at, notes)
@mcp.tool()
def tasks_list(owner: str | None = None, due_before: str | None = None) -> list[dict]:
"""List live tasks. Every task here is open by definition (done/cancelled tasks
are deleted). Ordered by deadline, soonest first, undated last.
Args:
owner: Filter by "agent" or "user". Omit for all.
due_before: ISO 8601; only tasks due before this. Useful for finding what's urgent.
"""
return tasks.list_tasks(owner, due_before)
@mcp.tool()
def tasks_update(task_id: int, content: str | None = None, due_at: str | None = None,
owner: str | None = None, notes: str | None = None) -> dict:
"""Update a task's fields. Only provided fields change. Use for refining content,
adjusting the deadline, reassigning owner, or appending notes. To finish or drop a
task, use complete_task / cancel_task instead."""
return tasks.update_task(task_id, content, due_at, owner, notes)
def _tasks_remove(task_id: int, context: str, state: str) -> dict:
result = tasks.remove_task(task_id)
if "error" in result:
return result
deleted = result["deleted"]
log_content = f"Task '{deleted['content']}' has been {state}. {context}"
log_entry = diary.log(log_content)
return {"deleted": deleted, "logged": log_entry}
@mcp.tool()
def tasks_complete(task_id: int, context: str) -> dict:
"""Remove a finished task from the live list.
Args:
task_id: identifier of the task.
context: brief summary of the task completion.
"""
return _tasks_remove(task_id, context, 'completed')
@mcp.tool()
def tasks_cancel(task_id: int, context: str) -> dict:
"""Remove a no-longer-relevant task from the live list.
Args:
task_id: identifier of the task.
context: reason for the task cancellation.
"""
return _tasks_remove(task_id, context, 'cancelled')
@mcp.tool()
def tasks_nudged(task_id: int) -> dict:
"""Record that you've just reminded the user about a (user-owned) task, updating its
last_nudged_at. Call immediately after nudging, so you can throttle reminders. Check
last_nudged_at before nudging again; re-nudge more often as the deadline approaches,
sparingly for distant or undated tasks."""
return tasks.record_nudge(task_id)
if __name__ == "__main__":
run_migrations()
mcp.run()
|
