add tool calling, SQLite persistence, group chat, image vision, voice transcription
Major features: - OpenAI function calling with tool call loop (streaming SSE parsing) - Built-in tools: spawn_agent (async claude -p), agent_status, kill_agent, update_scratch, send_file - Script-based tool discovery: tools/ dir with --schema convention - Feishu todo management script (tools/manage_todo) - SQLite persistence: conversations, messages, config, scratch_area tables - Sliding window context (100 msgs, slide 50, auto-summarize) - Conversation summary generation via LLM on window slide - Group chat support with independent session contexts - Image understanding: multimodal vision input (base64 to API) - Voice transcription via faster-whisper Docker service - Configurable persona stored in DB - diag command for session diagnostics - System prompt restructured: persona + tool instructions separated - RUST_BACKTRACE=1 in service, clippy in deploy pipeline - .gitignore for config/state/db files
This commit is contained in:
Fam Zheng
187
tools/manage_todo
Executable file
187
tools/manage_todo
Executable file
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.11"
|
||||
# dependencies = ["requests"]
|
||||
# ///
|
||||
"""Feishu Bitable todo manager.
|
||||
|
||||
Usage:
|
||||
./fam-todo.py list-undone List open todos
|
||||
./fam-todo.py list-done List completed todos
|
||||
./fam-todo.py add <title> Add a new todo
|
||||
./fam-todo.py mark-done <record_id> Mark as done
|
||||
./fam-todo.py mark-undone <record_id> Mark as undone
|
||||
./fam-todo.py --schema Print tool schema JSON
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import requests
|
||||
|
||||
APP_ID = "cli_a7f042e93d385013"
|
||||
APP_SECRET = "ht4FCjQ8JJ65ZPUWlff6ldFBmaP0mxqY"
|
||||
APP_TOKEN = "SSoGbmGFoazJkUs7bbfcaSG8n7f"
|
||||
TABLE_ID = "tblIA2biceDpvr35"
|
||||
BASE_URL = "https://open.feishu.cn/open-apis"
|
||||
|
||||
ACTIONS = ["list-undone", "list-done", "add", "mark-done", "mark-undone"]
|
||||
|
||||
SCHEMA = {
|
||||
"name": "fam_todo",
|
||||
"description": "管理 Fam 的飞书待办事项表格。",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"action": {
|
||||
"type": "string",
|
||||
"enum": ACTIONS,
|
||||
"description": "操作类型",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "待办标题 (add 时必填)",
|
||||
},
|
||||
"record_id": {
|
||||
"type": "string",
|
||||
"description": "记录ID (mark-done/mark-undone 时必填)",
|
||||
},
|
||||
},
|
||||
"required": ["action"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def get_token():
|
||||
r = requests.post(
|
||||
f"{BASE_URL}/auth/v3/tenant_access_token/internal/",
|
||||
json={"app_id": APP_ID, "app_secret": APP_SECRET},
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()["tenant_access_token"]
|
||||
|
||||
|
||||
def headers():
|
||||
return {"Authorization": f"Bearer {get_token()}", "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def api(method, path, **kwargs):
|
||||
url = f"{BASE_URL}/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}{path}"
|
||||
r = requests.request(method, url, headers=headers(), **kwargs)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
|
||||
def format_field(v):
|
||||
if isinstance(v, list):
|
||||
return "".join(
|
||||
seg.get("text", str(seg)) if isinstance(seg, dict) else str(seg)
|
||||
for seg in v
|
||||
)
|
||||
return str(v)
|
||||
|
||||
|
||||
def list_records(done_filter):
|
||||
"""List records. done_filter: True=done only, False=undone only."""
|
||||
data = api("GET", "/records", params={"page_size": 500})
|
||||
items = data.get("data", {}).get("items", [])
|
||||
if not items:
|
||||
return "No records found."
|
||||
|
||||
lines = []
|
||||
for item in items:
|
||||
fields = item.get("fields", {})
|
||||
is_done = bool(fields.get("Done"))
|
||||
if is_done != done_filter:
|
||||
continue
|
||||
|
||||
rid = item["record_id"]
|
||||
title = format_field(fields.get("Item", ""))
|
||||
priority = fields.get("Priority", "")
|
||||
notes = format_field(fields.get("Notes", ""))
|
||||
|
||||
parts = [f"[{rid}] {title}"]
|
||||
if priority:
|
||||
parts.append(f" P: {priority}")
|
||||
if notes:
|
||||
preview = notes[:80].replace("\n", " ")
|
||||
parts.append(f" Note: {preview}")
|
||||
lines.append("\n".join(parts))
|
||||
|
||||
if not lines:
|
||||
label = "completed" if done_filter else "open"
|
||||
return f"No {label} todos."
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def add_record(title):
|
||||
data = api("POST", "/records", json={"fields": {"Item": title}})
|
||||
rid = data.get("data", {}).get("record", {}).get("record_id", "?")
|
||||
return f"Added [{rid}]: {title}"
|
||||
|
||||
|
||||
def mark_done(record_id):
|
||||
api("PUT", f"/records/{record_id}", json={"fields": {"Done": True}})
|
||||
return f"Marked [{record_id}] as done"
|
||||
|
||||
|
||||
def mark_undone(record_id):
|
||||
api("PUT", f"/records/{record_id}", json={"fields": {"Done": False}})
|
||||
return f"Marked [{record_id}] as undone"
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h"):
|
||||
print(__doc__.strip())
|
||||
sys.exit(0)
|
||||
|
||||
if sys.argv[1] == "--schema":
|
||||
print(json.dumps(SCHEMA, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
|
||||
arg = sys.argv[1]
|
||||
if not arg.startswith("{"):
|
||||
args = {"action": arg}
|
||||
if len(sys.argv) > 2:
|
||||
args["title"] = " ".join(sys.argv[2:])
|
||||
args["record_id"] = sys.argv[2] # also set record_id for mark-*
|
||||
else:
|
||||
try:
|
||||
args = json.loads(arg)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Invalid JSON: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
action = args.get("action", "")
|
||||
try:
|
||||
if action == "list-undone":
|
||||
print(list_records(done_filter=False))
|
||||
elif action == "list-done":
|
||||
print(list_records(done_filter=True))
|
||||
elif action == "add":
|
||||
title = args.get("title", "")
|
||||
if not title:
|
||||
print("Error: title is required")
|
||||
sys.exit(1)
|
||||
print(add_record(title))
|
||||
elif action == "mark-done":
|
||||
rid = args.get("record_id", "")
|
||||
if not rid:
|
||||
print("Error: record_id is required")
|
||||
sys.exit(1)
|
||||
print(mark_done(rid))
|
||||
elif action == "mark-undone":
|
||||
rid = args.get("record_id", "")
|
||||
if not rid:
|
||||
print("Error: record_id is required")
|
||||
sys.exit(1)
|
||||
print(mark_undone(rid))
|
||||
else:
|
||||
print(f"Unknown action: {action}. Valid: {', '.join(ACTIONS)}")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user