@ -12,8 +12,9 @@ to use oauth. Review the full docs above and reach out to nla@zapier.com for
developer support .
"""
import json
from typing import Dict, List , Optional
from typing import Any, Dict, List , Optional
import aiohttp
import requests
from pydantic import BaseModel , Extra , root_validator
from requests import Request , Session
@ -49,36 +50,63 @@ class ZapierNLAWrapper(BaseModel):
extra = Extra . forbid
def _get_session ( self ) - > Session :
session = requests . Session ( )
session . headers . update (
{
" Accept " : " application/json " ,
" Content-Type " : " application/json " ,
}
)
def _format_headers ( self ) - > Dict [ str , str ] :
""" Format headers for requests. """
headers = {
" Accept " : " application/json " ,
" Content-Type " : " application/json " ,
}
if self . zapier_nla_oauth_access_token :
session. headers. update (
headers. update (
{ " Authorization " : f " Bearer { self . zapier_nla_oauth_access_token } " }
)
else :
session . params = { " api_key " : self . zapier_nla_api_key }
headers . update ( { " X-API-Key " : self . zapier_nla_api_key } )
return headers
def _get_session ( self ) - > Session :
session = requests . Session ( )
session . headers . update ( self . _format_headers ( ) )
return session
def _get_action_request (
self , action_id : str , instructions : str , params : Optional [ Dict ] = None
) - > Request :
async def _arequest ( self , method : str , url : str , * * kwargs : Any ) - > Dict [ str , Any ] :
""" Make an async request. """
async with aiohttp . ClientSession ( headers = self . _format_headers ( ) ) as session :
async with session . request ( method , url , * * kwargs ) as response :
response . raise_for_status ( )
return await response . json ( )
def _create_action_payload ( # type: ignore[no-untyped-def]
self , instructions : str , params : Optional [ Dict ] = None , preview_only = False
) - > Dict :
""" Create a payload for an action. """
data = params if params else { }
data . update (
{
" instructions " : instructions ,
}
)
if preview_only :
data . update ( { " preview_only " : True } )
return data
def _create_action_url ( self , action_id : str ) - > str :
""" Create a url for an action. """
return self . zapier_nla_api_base + f " exposed/ { action_id } /execute/ "
def _create_action_request ( # type: ignore[no-untyped-def]
self ,
action_id : str ,
instructions : str ,
params : Optional [ Dict ] = None ,
preview_only = False ,
) - > Request :
data = self . _create_action_payload ( instructions , params , preview_only )
return Request (
" POST " ,
self . zapier_nla_api_base + f " exposed/ { action_id } /execute/ " ,
self . _create_action_url( action_id ) ,
json = data ,
)
@ -107,6 +135,28 @@ class ZapierNLAWrapper(BaseModel):
return values
async def alist ( self ) - > List [ Dict ] :
""" Returns a list of all exposed (enabled) actions associated with
current user ( associated with the set api_key ) . Change your exposed
actions here : https : / / nla . zapier . com / demo / start /
The return list can be empty if no actions exposed . Else will contain
a list of action objects :
[ {
" id " : str ,
" description " : str ,
" params " : Dict [ str , str ]
} ]
` params ` will always contain an ` instructions ` key , the only required
param . All others optional and if provided will override any AI guesses
( see " understanding the AI guessing flow " here :
https : / / nla . zapier . com / api / v1 / docs )
"""
response = await self . _arequest ( " GET " , self . zapier_nla_api_base + " exposed/ " )
return response [ " results " ]
def list ( self ) - > List [ Dict ] :
""" Returns a list of all exposed (enabled) actions associated with
current user ( associated with the set api_key ) . Change your exposed
@ -157,11 +207,29 @@ class ZapierNLAWrapper(BaseModel):
call .
"""
session = self . _get_session ( )
request = self . _ get _action_request( action_id , instructions , params )
request = self . _ create _action_request( action_id , instructions , params )
response = session . send ( session . prepare_request ( request ) )
response . raise_for_status ( )
return response . json ( ) [ " result " ]
async def arun (
self , action_id : str , instructions : str , params : Optional [ Dict ] = None
) - > Dict :
""" Executes an action that is identified by action_id, must be exposed
( enabled ) by the current user ( associated with the set api_key ) . Change
your exposed actions here : https : / / nla . zapier . com / demo / start /
The return JSON is guaranteed to be less than ~ 500 words ( 350
tokens ) making it safe to inject into the prompt of another LLM
call .
"""
response = await self . _arequest (
" POST " ,
self . _create_action_url ( action_id ) ,
json = self . _create_action_payload ( instructions , params ) ,
)
return response [ " result " ]
def preview (
self , action_id : str , instructions : str , params : Optional [ Dict ] = None
) - > Dict :
@ -171,25 +239,58 @@ class ZapierNLAWrapper(BaseModel):
session = self . _get_session ( )
params = params if params else { }
params . update ( { " preview_only " : True } )
request = self . _ get _action_request( action_id , instructions , params )
request = self . _ create _action_request( action_id , instructions , params , True )
response = session . send ( session . prepare_request ( request ) )
response . raise_for_status ( )
return response . json ( ) [ " input_params " ]
async def apreview (
self , action_id : str , instructions : str , params : Optional [ Dict ] = None
) - > Dict :
""" Same as run, but instead of actually executing the action, will
instead return a preview of params that have been guessed by the AI in
case you need to explicitly review before executing . """
response = await self . _arequest (
" POST " ,
self . _create_action_url ( action_id ) ,
json = self . _create_action_payload ( instructions , params , preview_only = True ) ,
)
return response [ " result " ]
def run_as_str ( self , * args , * * kwargs ) - > str : # type: ignore[no-untyped-def]
""" Same as run, but returns a stringified version of the JSON for
insertting back into an LLM . """
data = self . run ( * args , * * kwargs )
return json . dumps ( data )
async def arun_as_str ( self , * args , * * kwargs ) - > str : # type: ignore[no-untyped-def]
""" Same as run, but returns a stringified version of the JSON for
insertting back into an LLM . """
data = await self . arun ( * args , * * kwargs )
return json . dumps ( data )
def preview_as_str ( self , * args , * * kwargs ) - > str : # type: ignore[no-untyped-def]
""" Same as preview, but returns a stringified version of the JSON for
insertting back into an LLM . """
data = self . preview ( * args , * * kwargs )
return json . dumps ( data )
async def apreview_as_str ( # type: ignore[no-untyped-def]
self , * args , * * kwargs
) - > str :
""" Same as preview, but returns a stringified version of the JSON for
insertting back into an LLM . """
data = await self . apreview ( * args , * * kwargs )
return json . dumps ( data )
def list_as_str ( self ) - > str : # type: ignore[no-untyped-def]
""" Same as list, but returns a stringified version of the JSON for
insertting back into an LLM . """
actions = self . list ( )
return json . dumps ( actions )
async def alist_as_str ( self ) - > str : # type: ignore[no-untyped-def]
""" Same as list, but returns a stringified version of the JSON for
insertting back into an LLM . """
actions = await self . alist ( )
return json . dumps ( actions )