Adding an Allowlist to Your Agent

This is part of the Developer Tutorial, but for clarity it's been added here as well.

Allowlists make it easy for your to control who can send requests to our agent. This is useful for both preventing spam as well as ensuring that, if you wish, only your customers can access your agent.

This all comes down to the fact that our xmtp-service.js sends the address of the message sender in the API request to the FastAPI app. This is sent as a header argument called "Sender," so to access the header we need to modify our route to accept a Request model:

Now we can use the header, but we need to impose some rules on it before we actually write our whitelisting logic. We don't want the sender address to be invalid.

Great, now we can be sure that the sender variable is going to actually have a valid sender address. I am now going to create a whitelisting function called check_whitelist that uses neynar.com APIs to make this agent only available to addresses which have been associated with a farcaster.xyz account.

We need to import time as a module and set a variable neynar_key, then we can use this check_whitelist function:

app = FastAPI()
load_dotenv()
neynar_key = os.environ.get("NEYNAR_SQL_API_KEY")

"""
Check whether a given address is registered on Farcaster or not.
If yes, then they are approved for the whitelist.

Parameters:
address: Ethereum address starting with 0x

Returns:
true or false
"""
def check_whitelist(address: str) -> bool:
    url = 'https://data.hubs.neynar.com/api/queries/257/results'
    params = {'api_key': neynar_key}
    payload = {
        "max_age": 1800,
        "parameters": {
            "address": address.strip().lower()
        }
    }
    headers = {'Content-Type': 'application/json'}
    response = requests.post(url, params=params, headers=headers, json=payload).json()
    if "query_result" not in list(response.keys()):
        if "job" not in list(response.keys()):
            raise ValueError("Error while trying to find matches. Is your API key valid?")
        else:
            time.sleep(1)
            response = requests.post(url, params=params, headers=headers, json=payload).json()
            if "query_result" not in list(response.keys()):
                raise ValueError("Error while trying to find matches. Is your API key valid?")
                
    rows = response["query_result"]["data"]["rows"]
    return len(rows) > 0

Now before we go any further, I am going to my fly.io dashboard and set a new secret for NEYNAR_SQL_API_KEY.

Great. We can go back to our code and see where in our /entry route we can use our whitelist function.

I am going to add the logic right after our simple validation logic for the sender address, and have it throw a 400 error if the sender address is not in the whitelist:

@app.post("/entry")
def entry(entry: Entry, request: Request):
    # Verify that the sender header is present
    sender = request.headers.get('Sender')
    if not sender:
        raise HTTPException(status_code=400, detail="Sender header is required")
    elif not sender.startswith("0x"):
        raise HTTPException(status_code=400, detail="Sender address should start with 0x")
    elif len(sender) != 42:
        raise HTTPException(status_code=400, detail="Sender address must be a valid Ethereum address")
    
    # With the sender address, now you can do any sort of validation you want
    if not check_whitelist(sender):
        raise HTTPException(status_code=400, detail="Address not in whitelist")
    
    url = f"{os.environ.get('FIXIE_URL')}"

Now I am going to save this new main.py and deploy it to fly using fly deploy. Then it's time to test the app by sending a message from my XMTP address which is also an address I use for my farcaster account:

It looks like that worked. Now I will try messaging the agent using an address that is brand new:

You can see our message was successfully blocked using our new whitelist logic.