Skip to content

TinyClient

The TinyClient is the core class of the Python SDK used to programmatically interact with the Tiny Foundation API.

Source code in tiny_foundation/client.py 
  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
class TinyClient:
    def __init__(self, api_url: str = "https://tiny-foundation-api.spikinglabs.com", supabase_url: Optional[str] = None, supabase_key: Optional[str] = None):
        self.api_url = api_url.rstrip("/")
        self.supabase_url = supabase_url or os.environ.get("SUPABASE_URL")
        self.supabase_key = supabase_key or os.environ.get("SUPABASE_ANON_KEY")

        self.supabase: Optional[Client] = None
        if self.supabase_url and self.supabase_key:
            self.supabase = create_client(self.supabase_url, self.supabase_key)

    def login(self, email: str, password: str) -> Credentials:
        if not self.supabase:
            raise ValueError("SUPABASE_URL and SUPABASE_ANON_KEY must be configured in environment.")

        response = self.supabase.auth.sign_in_with_password({
            "email": email,
            "password": password
        })

        if not response.session:
            raise ValueError("Login failed")

        self.credentials = Credentials(
            access_token=response.session.access_token,
            refresh_token=response.session.refresh_token,
            email=response.user.email
        )
        save_credentials(self.credentials)
        return self.credentials

    def push_document(self, workspace_id: str, access_token: str, title: str, content: str):
        url = f"{self.api_url}/api/workspaces/{workspace_id}/documents"
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }
        payload = {
            "title": title,
            "content": content
        }

        # In production, ensure verify=True for security. It's true by default in httpx.
        response = httpx.post(url, headers=headers, json=payload, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def list_workspaces(self, access_token: str) -> list:
        url = f"{self.api_url}/api/workspaces"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = httpx.get(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def export_workspace(self, workspace_id: str, access_token: str) -> list:
        url = f"{self.api_url}/api/workspaces/{workspace_id}/export"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = httpx.get(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def list_documents(self, workspace_id: str, access_token: str) -> list:
        url = f"{self.api_url}/api/workspaces/{workspace_id}/documents"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = httpx.get(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def approve_document(self, workspace_id: str, access_token: str, document_id: str, message: str = ""):
        url = f"{self.api_url}/api/workspaces/{workspace_id}/documents/{document_id}/approve"
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }
        payload = {"message": message}
        response = httpx.post(url, headers=headers, json=payload, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def trigger_graphify(self, workspace_id: str, access_token: str):
        """
        Triggers the backend pipeline to parse all approved documents in a workspace and build a knowledge graph.

        Args:
            workspace_id (str): The ID of the workspace.
            access_token (str): The user's authentication token.

        Returns:
            dict: Status of the extraction trigger.
        """
        url = f"{self.api_url}/api/workspaces/{workspace_id}/graphify/trigger"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = httpx.post(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()

    def get_graph(self, workspace_id: str, access_token: str) -> dict:
        """
        Retrieves the latest knowledge graph built for the workspace.

        Args:
            workspace_id (str): The ID of the workspace.
            access_token (str): The user's authentication token.

        Returns:
            dict: The graph containing `nodes` and `edges` lists. If no graph exists, returns an `error` key.
        """
        url = f"{self.api_url}/api/workspaces/{workspace_id}/graph"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = httpx.get(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()

get_graph(workspace_id, access_token)

Retrieves the latest knowledge graph built for the workspace.

Parameters:

Name Type Description Default
workspace_id str

The ID of the workspace.

 required
access_token str

The user's authentication token.

 required

Returns:

Name Type Description
dict dict

The graph containing nodes and edges lists. If no graph exists, returns an error key.
Source code in tiny_foundation/client.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get_graph(self, workspace_id: str, access_token: str) -> dict:
    """
    Retrieves the latest knowledge graph built for the workspace.

    Args:
        workspace_id (str): The ID of the workspace.
        access_token (str): The user's authentication token.

    Returns:
        dict: The graph containing `nodes` and `edges` lists. If no graph exists, returns an `error` key.
    """
    url = f"{self.api_url}/api/workspaces/{workspace_id}/graph"
    headers = {"Authorization": f"Bearer {access_token}"}
    response = httpx.get(url, headers=headers, timeout=10.0)
    response.raise_for_status()
    return response.json()

trigger_graphify(workspace_id, access_token)

Triggers the backend pipeline to parse all approved documents in a workspace and build a knowledge graph.

Parameters:

Name Type Description Default
workspace_id str

The ID of the workspace.

 required
access_token str

The user's authentication token.

 required

Returns:

Name Type Description
dict

Status of the extraction trigger.
Source code in tiny_foundation/client.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def trigger_graphify(self, workspace_id: str, access_token: str):
    """
    Triggers the backend pipeline to parse all approved documents in a workspace and build a knowledge graph.

    Args:
        workspace_id (str): The ID of the workspace.
        access_token (str): The user's authentication token.

    Returns:
        dict: Status of the extraction trigger.
    """
    url = f"{self.api_url}/api/workspaces/{workspace_id}/graphify/trigger"
    headers = {"Authorization": f"Bearer {access_token}"}
    response = httpx.post(url, headers=headers, timeout=10.0)
    response.raise_for_status()
    return response.json()