"""AgentForms API client — thin wrapper over /api/v2/.""" from __future__ import annotations from typing import Any, Literal, Optional import requests from pydantic import BaseModel from .types import ApiKey, FieldDefinition, Form, Submission PERMISSIONS = Literal["read_forms", "write_forms", "read_submissions", "delete_forms"] class AgentFormsError(Exception): """Raised on API errors.""" def __init__(self, message: str, status_code: int = 0, response: Any = None): super().__init__(message) self.status_code = status_code self.response = response class AgentForms: """Client for the AgentForms Agent API. Usage: from agentforms import AgentForms af = AgentForms(api_key="afk_live_...") # Create a form form = af.forms.create( name="Customer Feedback", fields=[ {"name": "rating", "label": "Rating", "type": "select", "options": ["1", "2", "3", "4", "5"], "required": True}, {"name": "feedback", "label": "Feedback", "type": "textarea"}, ], ) print(form.share_url) # Or generate with AI form = af.forms.generate(prompt="Build me a job application form for a restaurant") print(form.share_url) """ def __init__( self, api_key: str, base_url: str = "https://agentforms.io/api/v2", timeout: int = 30, ): self.api_key = api_key self.base_url = base_url.rstrip("/") self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json", }) self.timeout = timeout # Sub-clients self.forms = FormsClient(self) self.keys = KeysClient(self) self.submissions = SubmissionsClient(self) def _request(self, method: str, path: str, **kwargs) -> dict: url = f"{self.base_url}/{path.lstrip('/')}" resp = self._session.request(method, url, timeout=self.timeout, **kwargs) if resp.status_code >= 400: try: err = resp.json() message = err.get("error", resp.text) except Exception: message = resp.text raise AgentFormsError(message, resp.status_code, resp) if resp.status_code == 204: return {} return resp.json() class FormsClient: """Form operations: create, list, generate, update, delete.""" def __init__(self, client: AgentForms): self._c = client def list(self, limit: int = 50, offset: int = 0) -> dict: """List all forms.""" return self._c._request("GET", "/forms", params={"limit", "offset"}) def create( self, name: str, fields: list[dict | FieldDefinition], metadata: Optional[dict] = None, ) -> Form: """Create a new form programmatically. Args: name: Form display name fields: List of field definitions (dicts or FieldDefinition objects) metadata: Optional JSON metadata attached to the form Returns: Form object with token, URLs, and field definitions. """ field_list = [] for f in fields: if isinstance(f, BaseModel): field_list.append(f.model_dump(exclude_unset=True)) else: field_list.append(f) data: dict[str, Any] = {"name": name, "fields": field_list} if metadata: data["metadata"] = metadata result = self._c._request("POST", "/forms", json=data) return Form(**result) def generate( self, prompt: str, form_name: Optional[str] = None, ) -> Form: """Generate a form from natural language using AI. Tier-gated (Starter+). Args: prompt: Natural language description (e.g., "A survey for event feedback") form_name: Override the AI-generated form name Returns: Form object with AI-generated fields. """ data: dict[str, Any] = {"prompt": prompt} if form_name: data["form_name"] = form_name result = self._c._request("POST", "/forms/generate", json=data) return Form(**result) def get(self, token: str) -> dict: """Get form details by token.""" return self._c._request("GET", f"/forms/{token}") def update_fields(self, token: str, fields: list[dict | FieldDefinition]) -> dict: """Replace all fields on a form.""" field_list = [] for f in fields: if isinstance(f, BaseModel): field_list.append(f.model_dump(exclude_unset=True)) else: field_list.append(f) return self._c._request("PUT", f"/forms/{token}/fields", json={"fields": field_list}) def delete(self, token: str) -> dict: """Delete a form and all submissions.""" return self._c._request("DELETE", f"/forms/{token}") def config(self, token: str) -> dict: """Get public form config (no auth needed — uses session headers anyway).""" return self._c._request("GET", f"/forms/{token}/config") class SubmissionsClient: """Submission operations: list, get.""" def __init__(self, client: AgentForms): self._c = client def list( self, form_token: str, limit: int = 50, offset: int = 0, ) -> dict: """List submissions for a form.""" result = self._c._request( "GET", f"/forms/{form_token}/submissions", params={"limit", "offset"}, ) # Wrap submissions in models submissions = [] for sub in result.get("submissions", []): submissions.append(Submission(**sub)) result["submissions"] = submissions return result def get(self, form_token: str, submission_id: int) -> Submission: """Get a single submission.""" result = self._c._request("GET", f"/forms/{form_token}/submissions/{submission_id}") return Submission(**result) class KeysClient: """API key management.""" def __init__(self, client: AgentForms): self._c = client def list(self) -> list[ApiKey]: """List all API keys.""" result = self._c._request("GET", "/keys") return [ApiKey(**k) for k in result.get("keys", [])] def create( self, name: str, permissions: Optional[list[PERMISSIONS]] = None, ) -> dict: """Create a new API key. Args: name: Human-readable name permissions: List of permissions (defaults to all if omitted) Warning: The full_key is only returned once. Store it securely. """ data: dict[str, Any] = {"name": name} if permissions: perms = {p: True for p in permissions} data["permissions"] = perms return self._c._request("POST", "/keys", json=data) def revoke(self, key_id: int) -> dict: """Revoke an API key.""" return self._c._request("DELETE", f"/keys/{key_id}")