359cb24d9a
- client.py: async httpx wrapper, FableAPIError, stream_get() SSE generator, singleton init_client()/get_client() and _reset_client() for tests - tools/: notes, tasks, projects, milestones, search, chat — thin async functions that accept FableClient and call Fable REST endpoints - server.py: FastMCP entry point with 20 tools registered via @mcp.tool(), each opening a fresh FableClient context per call; validates env vars at startup - tests: 34 tests covering client HTTP behaviour, error handling, singleton, SSE streaming, and all tool modules Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
"""MCP tools for Fable notes."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fable_mcp.client import FableClient
|
|
|
|
|
|
async def list_notes(
|
|
client: FableClient,
|
|
*,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
tag: str | None = None,
|
|
search: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""List notes with optional filtering."""
|
|
params: dict[str, Any] = {"limit": limit, "offset": offset}
|
|
if tag:
|
|
params["tag"] = tag
|
|
if search:
|
|
params["search"] = search
|
|
return await client.get("/api/notes", params=params)
|
|
|
|
|
|
async def get_note(client: FableClient, *, note_id: int) -> dict[str, Any]:
|
|
"""Fetch a single note by ID."""
|
|
return await client.get(f"/api/notes/{note_id}")
|
|
|
|
|
|
async def create_note(
|
|
client: FableClient,
|
|
*,
|
|
title: str,
|
|
body: str = "",
|
|
tags: list[str] | None = None,
|
|
project_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new note."""
|
|
payload: dict[str, Any] = {"title": title, "body": body}
|
|
if tags is not None:
|
|
payload["tags"] = tags
|
|
if project_id is not None:
|
|
payload["project_id"] = project_id
|
|
return await client.post("/api/notes", json=payload)
|
|
|
|
|
|
async def update_note(
|
|
client: FableClient,
|
|
*,
|
|
note_id: int,
|
|
title: str | None = None,
|
|
body: str | None = None,
|
|
tags: list[str] | None = None,
|
|
project_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update an existing note."""
|
|
payload: dict[str, Any] = {}
|
|
if title is not None:
|
|
payload["title"] = title
|
|
if body is not None:
|
|
payload["body"] = body
|
|
if tags is not None:
|
|
payload["tags"] = tags
|
|
if project_id is not None:
|
|
payload["project_id"] = project_id
|
|
return await client.patch(f"/api/notes/{note_id}", json=payload)
|
|
|
|
|
|
async def delete_note(client: FableClient, *, note_id: int) -> None:
|
|
"""Delete a note by ID."""
|
|
await client.delete(f"/api/notes/{note_id}")
|