Files
ObsiGate/backend/tools/schemas.py
T
bruno 6a58a59a11
CI / lint (push) Successful in 1m37s
CI / security (push) Successful in 1m1s
CI / test (push) Successful in 3m26s
CI / build (push) Successful in 1m44s
CI / e2e (push) Successful in 11m1s
feat(ai): ecosysteme d'outils phase 2 - recherche a cle, cache/retry, Playwright, crawl, Gitea/GitHub, documents (#92)
2026-09-17 11:52:03 -04:00

346 lines
14 KiB
Python

"""Pydantic input/output models for the AI tool layer.
Input models double as JSON Schemas advertised to LLMs (via
``model_json_schema``) and as validation for arguments received over MCP.
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, Field
class ListVaultsInput(BaseModel):
"""No parameters — lists the vaults the caller can access."""
class ListDirectoryInput(BaseModel):
"""Browse a directory inside a vault."""
vault: str = Field(..., description="Vault name")
path: str = Field("", description="Vault-relative directory path (empty = vault root)")
class ReadFileInput(BaseModel):
"""Read a text file's content from a vault."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
class SearchFulltextInput(BaseModel):
"""Full-text search across one or all accessible vaults."""
q: str = Field(..., min_length=1, description="Search query")
vault: str = Field("all", description="Vault name or 'all'")
tag: str | None = Field(None, description="Optional comma-separated tag filter")
limit: int = Field(50, ge=1, le=200, description="Maximum number of results")
class ListTagsInput(BaseModel):
"""List tags and their occurrence counts."""
vault: str | None = Field(None, description="Vault name or 'all' (default: all accessible)")
class ListAllFilesInput(BaseModel):
"""List every file of a vault (optionally under a subdirectory)."""
vault: str = Field(..., description="Vault name")
dir: str = Field("", description="Vault-relative directory path (empty = vault root)")
limit: int = Field(200, ge=1, le=2000, description="Maximum number of files")
recursive: bool = Field(True, description="Recurse into subdirectories")
class ReadFileRawInput(BaseModel):
"""Read a text file's raw content (no HTML rendering)."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
class GetBacklinksInput(BaseModel):
"""List files linking to a target file via wikilinks."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the target file")
class ListBackupsInput(BaseModel):
"""List available backup versions of a file."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
class DiffBackupInput(BaseModel):
"""Unified diff between a backup version and another version or the current file."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
version: int = Field(..., description="Backup timestamp used as the old/left side")
compare_with: int | None = Field(
None,
description="Backup timestamp used as the new/right side (omit to compare with the current file)",
)
class GetGraphInput(BaseModel):
"""Graph data (nodes and edges) for a vault or directory."""
vault: str = Field(..., description="Vault name")
path: str = Field("", description="Vault-relative directory path to focus on (empty = root)")
depth: int = Field(1, ge=0, le=3, description="Expansion depth")
scope: str = Field("directory", description="'directory' for a subtree, 'full' for the whole vault")
tag: str = Field("", description="Only include files carrying this tag")
class SearchAdvancedInput(BaseModel):
"""Advanced full-text search (TF-IDF, facets, operators)."""
q: str = Field("", description="Query (supports tag:, vault:, title:, path:, ext: operators)")
vault: str = Field("all", description="Vault name or 'all'")
tag: str | None = Field(None, description="Optional comma-separated tag filter")
limit: int = Field(50, ge=1, le=200, description="Maximum number of results")
offset: int = Field(0, ge=0, description="Pagination offset")
sort: str = Field("relevance", description="Sort by 'relevance' or 'modified'")
case_sensitive: bool = Field(False, description="Match case")
whole_word: bool = Field(False, description="Match whole words only")
regex: bool = Field(False, description="Treat query as a regular expression")
include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
created: str | None = Field(None, description="Created date filter (>date, <date, date..date)")
modified: str | None = Field(None, description="Modified date filter (>date, <date, date..date, <Nd)")
size: str | None = Field(None, description="Size filter (>size, <size, size..size, e.g. >1MB)")
class SearchPathsInput(BaseModel):
"""Search files and directories by path substring."""
q: str = Field(..., min_length=1, description="Path substring to search for")
vault: str = Field("all", description="Vault name or 'all'")
class SuggestTagsInput(BaseModel):
"""Suggest tags matching a prefix."""
q: str = Field(..., min_length=1, description="Tag prefix (with or without leading '#')")
vault: str = Field("all", description="Vault name or 'all'")
limit: int = Field(10, ge=1, le=50, description="Maximum number of suggestions")
class ListRecentInput(BaseModel):
"""List recently opened (or modified) files."""
vault: str | None = Field(None, description="Optional single-vault filter")
limit: int = Field(20, ge=1, le=200, description="Maximum number of files")
mode: str = Field("opened", description="'opened' (history) or 'modified'")
# ── D. Mutations ───────────────────────────────────────────────────────────
class CreateFileInput(BaseModel):
"""Create a new text file in a vault."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the new file")
content: str = Field("", description="Initial file content")
class CreateDirectoryInput(BaseModel):
"""Create a new directory in a vault."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the new directory")
class EditFileInput(BaseModel):
"""Overwrite the full content of an existing file."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
content: str = Field(..., description="New full content of the file")
class AppendToFileInput(BaseModel):
"""Append text to the end of an existing file."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
content: str = Field(..., description="Text to append")
class RenameFileInput(BaseModel):
"""Rename a file in place (same parent directory)."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Current vault-relative file path")
new_name: str = Field(..., description="New file name (no directory separator)")
class RenameDirectoryInput(BaseModel):
"""Rename a directory in place (same parent directory)."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Current vault-relative directory path")
new_name: str = Field(..., description="New directory name (no directory separator)")
class MovePathInput(BaseModel):
"""Move a file or directory to another directory within the same vault."""
vault: str = Field(..., description="Vault name")
source_path: str = Field(..., description="Current vault-relative path of the file/directory")
destination_dir: str = Field("", description="Target directory path (empty = vault root)")
class ReplaceInFilesInput(BaseModel):
"""Find and replace text across vault files (destructive, dry-run by default)."""
find: str = Field(..., min_length=1, description="Text or pattern to search for")
replace: str = Field("", description="Replacement text")
vault: str = Field("all", description="Vault name or 'all'")
case_sensitive: bool = Field(False, description="Match case")
whole_word: bool = Field(False, description="Match whole words only")
regex: bool = Field(False, description="Treat 'find' as a regular expression")
include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
replace_all: bool = Field(False, description="Apply the replacement (otherwise only preview)")
dry_run: bool | None = Field(
None,
description="Preview matches without writing. Defaults to the opposite of replace_all.",
)
class DeleteFileInput(BaseModel):
"""Delete a file from a vault (destructive)."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
class DeleteDirectoryInput(BaseModel):
"""Delete a directory from a vault (destructive)."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative directory path")
recursive: bool = Field(True, description="Delete non-empty directories recursively")
class RestoreBackupInput(BaseModel):
"""Restore a file from one of its backup versions."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative file path")
version: int = Field(..., description="Backup timestamp to restore")
class WebSearchInput(BaseModel):
"""Search the public web through the configured meta-search engine."""
query: str = Field(..., description="Search terms")
max_results: int = Field(5, ge=1, le=10, description="Number of results to return")
category: str = Field("", description="Optional engine category: general, news, it, science")
language: str = Field("", description="Optional language code, e.g. 'fr'")
page: int = Field(1, ge=1, le=10, description="Result page number")
class FetchUrlInput(BaseModel):
"""Fetch one public web page and return its readable text."""
url: str = Field(..., description="Absolute http(s) URL of a public page")
render: bool = Field(
False,
description="Render JavaScript with the optional Playwright worker (dynamic SPA pages)",
)
class CrawlSiteInput(BaseModel):
"""Crawl a small public site (same-host only) and save a digest into a vault."""
url: str = Field(..., description="Absolute http(s) URL where the crawl starts")
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the digest file to write (.md)")
max_pages: int = Field(5, ge=1, le=20, description="Maximum number of pages to crawl")
class GitProviderInput(BaseModel):
"""Base fields for connected-source tools (Gitea / GitHub)."""
provider: str = Field(..., description="'gitea' (OBSIGATE_GITEA_URL) or 'github'")
repo: str = Field("", description="Optional 'owner/name' repository filter")
limit: int = Field(20, ge=1, le=50, description="Maximum number of entries")
class GitSearchIssuesInput(BaseModel):
"""Search issues/pull requests on a connected Gitea or GitHub instance."""
provider: str = Field(..., description="'gitea' or 'github'")
query: str = Field(..., min_length=1, description="Search keywords")
repo: str = Field("", description="Optional 'owner/name' scope (empty = instance-wide)")
state: str = Field("open", description="'open' or 'closed'")
limit: int = Field(10, ge=1, le=20, description="Maximum number of issues")
class GitGetFileInput(BaseModel):
"""Read a file from a connected Gitea or GitHub repository."""
provider: str = Field(..., description="'gitea' or 'github'")
repo: str = Field(..., description="'owner/name' repository")
path: str = Field(..., description="Repository-relative file path")
ref: str = Field("", description="Optional branch/tag/commit (empty = default branch)")
class SpreadsheetInput(BaseModel):
"""Create an .xlsx spreadsheet in a vault from rows of cells."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the file to write (.xlsx)")
rows: list[list[str | int | float | bool | None]] = Field(
..., description="Rows of cell values (first row = header)"
)
sheet_name: str = Field("Feuille1", description="Worksheet name")
overwrite: bool = Field(True, description="Replace an existing file (with backup)")
class DocxInput(BaseModel):
"""Create a .docx Word document in a vault from paragraphs."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the file to write (.docx)")
title: str = Field("", description="Optional document title (heading 1)")
paragraphs: list[str] = Field(..., description="Paragraph texts, in order")
overwrite: bool = Field(True, description="Replace an existing file (with backup)")
class CsvInput(BaseModel):
"""Create a .csv file in a vault from rows of cells."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the file to write (.csv)")
rows: list[list[str | int | float | bool | None]] = Field(
..., description="Rows of cell values (first row = header)"
)
delimiter: str = Field(",", description="Field separator (',' ';' '\\t')")
overwrite: bool = Field(True, description="Replace an existing file (with backup)")
class PdfInput(BaseModel):
"""Create a .pdf document in a vault from markdown-ish content."""
vault: str = Field(..., description="Vault name")
path: str = Field(..., description="Vault-relative path of the file to write (.pdf)")
title: str = Field("Document", description="Document title")
content: str = Field(..., description="Content (headings with #/##, then paragraphs)")
overwrite: bool = Field(True, description="Replace an existing file (with backup)")
class ToolResult(BaseModel):
"""Uniform result returned by :func:`backend.tools.registry.call_tool`."""
ok: bool = True
data: Any = None
error: str | None = None