Hey everyone. I've been seeing more folks deploying LangGraph graphs as standalone services with webhooks, which is fantastic. But a common question in the support channels is about securing those endpoints. You don't want just anyone hitting your graph's webhook and spinning up agents, right? 😅
Let's walk through a concrete pattern for adding JWT (JSON Web Token) validation to a graph's webhook handler. This ensures that incoming requests are from authenticated, authorized sources. We'll use a simple FastAPI setup as our webhook receiver, but the concepts apply elsewhere. The key is to intercept the request *before* it reaches the graph's invocation logic.
First, here's our baseline *insecure* webhook:
```python
from fastapi import FastAPI, Request
from my_graph import graph # Your compiled graph
app = FastAPI()
@app.post("/webhook")
async def handle_webhook(request: Request):
data = await request.json()
# Directly passing untrusted input to the graph
result = graph.invoke(data)
return result
```
To secure this, we'll add a dependency that validates a JWT from the `Authorization` header. We'll need `python-jose[cryptography]` and `passlib` for this example.
```python
from fastapi import FastAPI, Request, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from pydantic import BaseModel
from my_graph import graph
import os
SECRET_KEY = os.getenv("JWT_SECRET_KEY") # Keep this safe!
ALGORITHM = "HS256"
security = HTTPBearer()
class GraphInput(BaseModel):
user_id: str
query: str
def validate_jwt(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# Ensure the token has the expected scope/role
if payload.get("scope") != "graph_webhook":
raise HTTPException(status_code=403, detail="Invalid scope")
return payload # You can attach this to the request context
except JWTError:
raise HTTPException(status_code=401, detail="Invalid or expired token")
@app.post("/secure_webhook")
async def handle_secure_webhook(
request: GraphInput,
token_payload: dict = Depends(validate_jwt)
):
# The token is now validated. We can also use claims from the payload.
# For example, inject the authenticated user ID from the token, overriding any client input.
safe_input = request.dict()
safe_input["authenticated_user"] = token_payload["sub"]
# Now invoke the graph with the sanitized/safe input
result = graph.invoke(safe_input)
return result
```
**Key points to consider for production:**
* **Secret Management:** Never hardcode `SECRET_KEY`. Use environment variables or a secrets manager.
* **Input Sanitization:** The JWT validates the *caller*, but you must still validate/sanitize the graph's input. Notice how I'm taking the subject (`sub`) from the validated token and using that as the `authenticated_user`, ignoring any client-provided user ID. This prevents privilege escalation.
* **Claims are Powerful:** Use JWT claims (`scope`, `roles`, `graph_permissions`) to implement fine-grained access. Maybe only tokens with `"graph": "customer_support"` can invoke this particular graph.
* **Checkpointing Implications:** If your graph uses checkpointing with a remote store, the authenticated user claim should be part of the checkpoint metadata. This aids in audit trails and ensuring users can only resume their own sessions.
* **LangSmith:** Trace your JWT validation step as a separate LangSmith run, or at least ensure the authenticated user ID is attached as metadata to your graph runs for traceability.
This pattern adds a robust layer of authentication. It's a first step. From here, you might layer on rate limiting per `sub` claim, or more complex authorization logic before specific tool nodes execute.
Has anyone implemented a different auth pattern, like API keys with tool-level permissions? Would love to hear about other approaches.
- Tom (mod)
Nice start. The pre-invocation interception point is crucial, but I'd also validate the *structure* of the data payload after JWT auth. A valid token doesn't guarantee the input won't poison your graph's state.
You could have a malicious actor with a stolen token sending crafted payloads designed to, say, inject a poisoned system prompt into the graph's config for future runs. Your dependency should schematize the expected input.
Something like this after the JWT check:
```python
from pydantic import BaseModel, ValidationError
class WebhookInput(BaseModel):
message: str
thread_id: str
# ... other expected fields
# In your handler
try:
validated_input = WebhookInput(**data)
except ValidationError:
raise HTTPException(...)
```
It's a small extra layer that sanitizes the data shape before it ever touches your graph's context.
ak
Okay, this is super helpful for me because I'm just starting with webhooks. I've seen JWT mentioned a lot but never actually set it up myself. Can you explain what we put *inside* the dependency? Like, is it just a function that checks the token and returns something?
Also, where does the secret key for verifying the JWT usually come from? Do you store it in an environment variable?
Good call starting with the pre-invocation hook, that's where the actual security boundary gets defined. The `python-jose` and `passlib` suggestion is solid, but a lot of people forget you can skip that whole dependency tree if you're using a simple symmetric secret - PyJWT is leaner and has fewer transitive dependencies. The secret key absolutely should be an environment variable, ideally pulled from a vault at runtime.
But the real caveat is where you do the extraction. You need to verify the JWT's signature *before* you even start parsing its claims for things like `exp` or `sub`. I've seen people do it the other way around because the libraries make it tempting, and that's a trivial bypass if you're not careful.
Also, for these webhooks, consider using a short-lived JWT with a specific `aud` claim. That way, even if a token leaks from some other service, it can't be reused against your graph endpoint.
Escape artist, security consultant.
Great starting point with the pre-invocation hook. Using dependencies is the right move for FastAPI.
One small thing - you mentioned needing `python-jose` and `passlib`. For a symmetric secret, `PyJWT` is my go-to. It's lighter and the API's straightforward. Just `pip install PyJWT`. The dependency function ends up looking like this:
```python
import jwt
from fastapi import HTTPException, Depends
async def verify_token(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
raise HTTPException(401)
token = authorization[7:]
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload # or a user ID from the payload
except jwt.InvalidTokenError:
raise HTTPException(403)
```
Then your route just uses `Depends(verify_token)`. The secret key should absolutely come from an env var. What signing algorithm are you planning to use?
default deny
You're right about the pre-invocation hook being the right security boundary, but I've run into a nuance with the dependency approach. If your graph logic also needs something from the validated JWT payload (like a user ID for context), you end up passing it through the whole call chain.
I've started wrapping the graph invocation itself. The dependency handles the pure auth, and if that passes, you inject the validated claims into the request state. Then your handler can pull them out. Keeps the auth logic separate from your business logic.
```python
@app.post("/webhook")
async def handle_webhook(request: Request, user_claims = Depends(verify_token)):
data = await request.json()
# Now you have user_claims available here
result = graph.invoke({**data, "user": user_claims.get("sub")})
return result
```
That way the dependency doesn't need to know anything about the graph's input structure.
Self-host or die.
Injecting claims into request state is just shifting the problem. Now your graph logic is coupled to the web framework's state object.
Better to have the dependency return a simple object and pass it explicitly. The extra line of code is trivial, and you can test the graph without mocking a request context.
show me the proof, not the whitepaper