fastgws

fastgws builds async Python clients for Google APIs from Google’s discovery documents. Each service gets a small Python surface: resource groups become attributes, operations become awaitable methods, and responses come back as lightweight objects instead of raw JSON dictionaries.

Installation

Install from pypi:

$ pip install fastgws

Or install the latest development version from GitHub:

$ pip install git+https://github.com/answerdotai/fastgws.git

How to use

Import the service clients you want to use. Each client is built from Google’s discovery documents and exposes resource groups as Python attributes, so calls look like await drive.files.list(...) or await calendar.events.list(...).

from fastgws import Calendar, Docs, Drive, GMail, Places
from fastgws.auth import *

fastgws supports OAuth credentials and API keys. OAuth is the usual choice for Google Workspace APIs such as Gmail, Calendar, Drive, and Docs, because these APIs act on behalf of a user and require explicit scopes.

Authorization is deliberately separate from this library. Use gclientid once to create your Google Cloud project, consent configuration, and OAuth client, then use gclientid-auth whenever an account needs a token or additional scopes. fastgws only loads and refreshes those existing tokens; it never opens a browser or changes a grant.

API keys are useful for public Google APIs that support key-based access, such as Places. You can pass api_key=... directly or set GOOGLE_API_KEY or GWS_API_KEY in the environment.

Use a gclientid token

gclientid creates a Web OAuth client and one authorized-user token file per Google account. Authorize the account first, choosing a preset or explicit scopes:

gclientid-auth --account me@example.com --preset google-apps

Then pass the account name instead of constructing a path:

creds = await oauth_creds(account='me@example.com')

This loads $XDG_CONFIG_HOME/gclientid/oauth-token-me@example.com.json. The token records its granted scopes, so scopes is optional; when supplied, fastgws verifies that the token covers them. Access tokens are refreshed back into the same file while preserving mode 0600 and gclientid’s verified account metadata. Pass token_path= instead for an authorized-user file stored elsewhere.

For a client created by gclientid --internal, select its independent *-internal token without constructing a path:

creds = await oauth_creds(account="me@example.com", internal=True)

scopes = ['https://www.googleapis.com/auth/calendar', 'https://www.googleapis.com/auth/documents',
    'https://www.googleapis.com/auth/drive.readonly', 'https://www.googleapis.com/auth/gmail.readonly']
creds = await oauth_creds(account='me@example.com', scopes=scopes)

Auth complete

Responses are converted into lightweight Python objects. Known Google resource kinds get more specific classes such as FileList, Events, or Event; when a service does not provide enough schema information, fastgws falls back to the base GWSObject. Either way, fields are available as attributes as well as dictionary keys.

Use Docs to create a document, apply batch updates, and read the document back. The API accepts the same request dictionaries documented by Google, while fastgws handles auth, transport, and object conversion.

docs = Docs(creds=creds)
doc = await docs.documents.create(title='fastgws test doc')
doc

GWSObject(title='fastgws test doc', documentId='1ObmgD5GOA9zNZbUwCYeFZH8nUc_MKcJHKFqjPJGnkQs', body=1, documentStyle=11, namedStyles=1, tabs=1)

await docs.documents.batch_update(document_id=doc.documentId,
    requests=[{'insertText': {'location': {'index': 1},
        'text': 'Hello from fastgws\n'}}])

GWSObject(documentId='1ObmgD5GOA9zNZbUwCYeFZH8nUc_MKcJHKFqjPJGnkQs', replies=1, writeControl=1)

def doc_text(doc):
    return ''.join(e.textRun.content for b in doc.body.content if 'paragraph' in b for e in b.paragraph.elements if 'textRun' in e)

doc = await docs.documents.get(document_id=doc.documentId)
txt = doc_text(doc)
print(txt)

Hello from fastgws

Use Drive to search files and inspect metadata. This example returns a FileList, and its files collection contains file objects with attributes such as id, name, and mimeType.

drive = Drive(creds=creds)
fs = await drive.files.list(q="name contains 'fastgws' and trashed=false", page_size=10)
fs, fs.files[0]

(FileList(kind='drive#fileList', files=3),
 File(id='1ObmgD5GOA9zNZbUwCYeFZH8nUc_MKcJHKFqjPJGnkQs', name='fastgws test doc', mimeType='application/vnd.google-apps.document', kind='drive#file'))

Use Gmail to search messages with the Gmail query syntax. The result is still a Python object, so you can inspect message ids, thread ids, and any fields returned by the API without digging through raw JSON first.

gmail = GMail(creds=creds)
msgs = await gmail.users.messages.list(user_id='me', max_results=10)
msgs

GWSObject(messages=10)

List operations expose pages. The iterator forwards each nextPageToken as page_token and stops after the final page.

Every operation also exposes batch when its Google discovery document advertises a batch endpoint. Pass dictionaries containing the same arguments accepted by the operation; results preserve call order. fastgws uses Google’s recommended 50-call chunks by default (the protocol maximum is 100), and return_exceptions=True returns a structured APIError in the corresponding position instead of raising it.

messages = await gmail.users.messages.get.batch([
    dict(user_id='me', id=mid, format='minimal', fields='id,labelIds')
    for mid in message_ids
])

Ordinary and batched operations retry transient network failures, 429s, 5xx responses, and Google’s retryable 403 rate-limit reasons with truncated exponential backoff, jitter, and Retry-After support. A batch retries only its failed parts. Credentials refresh automatically after a 401. Clients request gzip responses by default; use Google’s global fields argument, as above, to request a partial response when the complete resource is unnecessary.

pages = gmail.users.messages.list.pages(user_id='me', max_results=10)
first_page = await anext(pages)
first_page

Use Calendar to create, update, delete, and search events.

calendar = Calendar(creds=creds)
event = await calendar.events.insert(calendar_id='primary', summary='fastgws test event', start={'dateTime': '2030-01-01T09:00:00Z'},
    end={'dateTime': '2030-01-01T09:30:00Z'})
event

Event(id='u99k6q861u35h6mrmejdrgc0gg', summary='fastgws test event', kind='calendar#event', creator=2, organizer=2, start=2, end=2, reminders=1)

events = await calendar.events.list(calendar_id='primary', q='fastgws test event',
    max_results=10, single_events=True, order_by='startTime')
events, events['items'][0]

(Events(summary='nc@answer.ai', kind='calendar#events', defaultReminders=1, items=1),
 Event(id='u99k6q861u35h6mrmejdrgc0gg', summary='fastgws test event', kind='calendar#event', creator=2, organizer=2, start=2, end=2, reminders=1))

await calendar.events.delete(calendar_id='primary', event_id=event.id)

''

Use API-key services the same way. Places can run with an API key instead of OAuth credentials, and this example asks Google to return only the fields needed to render a link.

from IPython.display import Markdown

places = Places()
res = await places.places.search_text(text_query='coffee near San Francisco',
    _headers={'X-Goog-FieldMask':'places.displayName,places.formattedAddress,places.location,places.googleMapsUri'})
p = res.places[0]
Markdown(f'[{p.displayName.text}]({p.googleMapsUri})')

fastgws can also create service clients dynamically from Google’s discovery index. If Google publishes a discovery document for a service, you can usually import that service by name, for example from fastgws import Sheets, then use it with the same creds, token, or api_key arguments shown above.

Services outside the central discovery index can be loaded from their own discovery URL. The discovery request uses the same credentials, token, API key, quota project, and custom headers as the resulting client.

drive = await GWSApi.from_discovery_url(
    'https://www.googleapis.com/discovery/v1/apis/drive/v3/rest',
    creds=creds)

Workspace Add-ons

WorkspaceAddons manages HTTP add-on deployments through the documented REST endpoints. It does not fetch the Workspace Add-ons discovery document. deploy creates or replaces one stable deployment ID, and ensure_installed installs it for the credentials’ user only when needed. After replacing a development deployment’s callback configuration, use reinstall so host applications pick up the change.

from fastgws import WorkspaceAddons

addons = WorkspaceAddons('my-project', creds)
auth = await addons.authorization()
deployment = await addons.deploy('dev', manifest)
status = await addons.ensure_installed('dev')

Create one client per tester because test installation belongs to the authenticated user.

Workspace administration

WorkspaceAdmin provides explicit user lifecycle operations over the Admin Directory and Enterprise License Manager APIs. Creation does not imply licensing: domains can auto-assign licences by organizational unit, or callers can use assign_license with the domain’s product SKU.

from fastgws import WorkspaceAdmin

admin = WorkspaceAdmin(creds)
user = await admin.create_user('new@example.com', 'New', 'User', password,
                               org_unit_path='/Internal')
await admin.assign_license(user.primaryEmail, 'workspace-sku-id')

Suspension, restoration, licence removal, and deletion are separate calls, so automated setup does not hide destructive lifecycle changes.

PySkill support

fastgws includes a PySkill for agents working inside solveit. Load fastgws.skill when a task needs access to Google Workspace or Google APIs through the base GWSApi client.

The skill exposes GWSApi, GWSObject, oauth_creds, and svc_acct_creds, and allows generated Google API operations through GWSOpFunc. Agents load an account’s existing gclientid token; authorization remains an explicit user action through gclientid-auth.

creds = await oauth_creds(account='me@example.com', scopes=['https://www.googleapis.com/auth/gmail.readonly'])
gmail = GWSApi('gmail', creds=creds)
msgs = await gmail.users.messages.list(user_id='me', max_results=10)