"""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}")