A simple asynchronous Python 3.6+ wrapper of ArchiSteamFarm IPC API
Powered by aiohttp, ASF_IPC is now fully asynchronous. Synchronous methods are no longer supported. ASF_IPC resolves APIs automatically from the new swagger API of ASF, so ASF_IPC always has the latest APIs data as long as ASF is updated.
- Python 3.6+
- aiohttp
pip3 install -U ASF_IPCThis example shows how to send a command to ASF:
import asyncio
from ASF import IPC
async def command(asf, cmd):
return await asf.Api.Command['command'].post(command=cmd)
async def main():
# The IPC initialization duration depends on the network
async with IPC(ipc='http://127.0.0.1:1242', password='YOUR IPC PASSWORD') as asf:
while True:
cmd = input('Enter a command: ')
resp = await command(asf, cmd)
if resp.success:
print(resp.result)
else:
print(f'Error: {resp.message}')
loop = asyncio.get_event_loop()
output = loop.run_until_complete(main())
loop.close()To get a list of all endpoints of ASF, open your web browser and visit the swagger page of your ASF instance (usually http://127.0.0.1:1242/swagger).
You can see many endpoints with their path, such as /Api/Bot/{botNames}, this endpoint in ASF_IPC is asf.Api.Bot['botNames'].
The replacing rule is simple: change /{text} to ['text'] and change / to .
Some more examples:
asf.Api.ASF # /Api/ASF
asf.Api.Command['command'] # /Api/Command/{command}
asf.Api.Bot['botNames'].Pause # /Api/Bot/{botNames}/Pause
asf.Api.WWW.GitHub.Releases # /Api/WWW/GitHub/Releases
asf.Api.WWW.GitHub.Releases['version'] # /Api/WWW/GitHub/Releases/{version}After your endpoint found, you want to send some data to this endpoint, you can use get(), post(), put(), delete() methods, these methods have optional arguments:
body(dict): the JSON request body.params(dict): the parameters in URL after a?, e.g. theparamsof/Api/WWW/GitHub/Releases?count=10is{'count': 10}
If you need to pass parameters in the path of the endpoint, e.g. {botName} in /Api/Bot/{botName}/Redeem, you can pass them as a keyword parameter of the method.
Some examples:
# POST /Api/Command/status%20asf
await asf.Api.Command['command'].post(command='status asf')
# GET /Api/WWW/GitHub/Releases?count=10
await asf.Api.WWW.GitHub.Releases.get(params={'count': 10})
# POST /Api/Bot/robot with json body {'BotConfig': ...}
await asf.Api.Bot['botName'].post(body={'BotConfig': ...}, botName='robot')After sending a request to the endpoint, we got a response object, which has 3 attributes:
Resultorresult(str): some data returned by ASF.Messageormessage(str): describes what happened with the request.Successorsuccess(bool): if the request has succeeded.
If ASF_IPC cannot give a value to some attributes, these attributes will be None or empty value.
Example for /Api/NLog:
import asyncio
from ASF import IPC
async def get_log(asf):
async for resp in asf.Api.NLog.ws(): # use ws() instead of get(), post()...
if resp.success:
print(resp.result)
async def main():
async with IPC(ipc='http://127.0.0.1:1242', password='YOUR IPC PASSWORD') as asf:
while True:
await get_log(asf)
loop = asyncio.get_event_loop()
output = loop.run_until_complete(main())
loop.close()