Python KeyError in Celery Task: 3 Causes (2026)

A Python KeyError when calling a Celery task usually means the worker received different keyword arguments than the task expected. Common 2026 causes: .delay() call with wrong kwarg names, the JSON serializer drops a complex object key, or task chaining passed an unexpected positional argument.

Minimal reproducer

# tasks.py
from celery import Celery
app = Celery('demo', broker='redis://localhost')

@app.task
def send_email(user_id, template):
    return f"Sending {template} to {user_id}"

# caller.py
send_email.delay(uid=42, template='welcome')
# Worker raises: TypeError: send_email() got unexpected kwarg 'uid'
# Or KeyError if you access kwargs['user_id'] in the task body

Fix 1: Match the task signature exactly

# Wrong
send_email.delay(uid=42, template='welcome')

# Right
send_email.delay(user_id=42, template='welcome')

# OR positional
send_email.delay(42, 'welcome')

Fix 2: Use named arguments with defaults

@app.task
def send_email(user_id, template='welcome', **extra):
    """Defensive task signature with **extra catches typos."""
    if 'uid' in extra:
        user_id = extra['uid']  # backwards-compat for callers
    return f"Sending {template} to {user_id}"

Fix 3: Switch serializer to pickle for complex objects

# celery_app.py
app = Celery('demo', broker='redis://localhost')
app.conf.task_serializer = 'json'  # default, safe but limited
# OR for complex objects:
app.conf.task_serializer = 'pickle'  # supports any Python object
app.conf.accept_content = ['pickle', 'json']

JSON serializer cannot serialize datetimes, sets, custom objects. They get dropped or stringified. Switch to pickle if you need to pass complex Python objects (with the caveat that pickle has security implications, never accept tasks from untrusted sources).

Fix 4: Type-safe tasks with Pydantic argument models

from pydantic import BaseModel

class EmailArgs(BaseModel):
    user_id: int
    template: str = 'welcome'

@app.task
def send_email(payload):
    args = EmailArgs(**payload)  # validates at task boundary
    return f"Sending {args.template} to {args.user_id}"

# caller
send_email.delay(payload={'user_id': 42, 'template': 'reset'})

Debugging Celery KeyError

• Set CELERY_TASK_TRACK_STARTED=True to log task starts.
• Run worker in foreground: celery -A myapp worker --loglevel=debug
• Inspect failed task in Flower (pip install flower; celery -A myapp flower) for full kwargs trace.
• Use task.apply() instead of .delay() during dev for synchronous execution.