	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Cloud Tasks - Async Job Processing

MCP Server with Cloud Tasks - Async Job Processing

Author: Venkata Sudhakar

Cloud Tasks is a fully managed service for asynchronous task execution. When an AI agent needs to trigger a long-running operation - such as generating a bulk report, sending a batch of emails, or processing an uploaded file - it should not block waiting for completion. Cloud Tasks lets the agent dispatch the job and move on immediately.

ShopMax India uses Cloud Tasks to handle background jobs like bulk invoice generation, stock replenishment notifications, and nightly data exports. The MCP server below gives ADK agents two tools - one to enqueue a job and another to check its status via a tracking record in Firestore.

The below example shows an MCP server that enqueues background jobs via Cloud Tasks and tracks their status, allowing agents to dispatch and monitor async work.

# Install: pip install mcp google-cloud-tasks google-cloud-firestore
import asyncio
import json
import uuid
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from google.cloud import tasks_v2, firestore

PROJECT = "shopmax-india"
LOCATION = "asia-south1"
QUEUE = "shopmax-jobs"
WORKER_URL = "https://worker.shopmax.in/process"

tasks_client = tasks_v2.CloudTasksClient()
db = firestore.Client(project=PROJECT)
app = Server("cloudtasks-mcp")

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="enqueue_job",
            description="Enqueue a background job via Cloud Tasks",
            inputSchema={
                "type": "object",
                "properties": {
                    "job_type": {"type": "string",
                        "description": "Job type: bulk_invoice, stock_alert, data_export"},
                    "payload": {"type": "object", "description": "Job-specific parameters"}
                },
                "required": ["job_type", "payload"]
            }
        ),
        types.Tool(
            name="check_job_status",
            description="Check the status of a previously enqueued background job",
            inputSchema={
                "type": "object",
                "properties": {
                    "job_id": {"type": "string"}
                },
                "required": ["job_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "enqueue_job":
        job_id = str(uuid.uuid4())
        payload = {"job_id": job_id, "job_type": arguments["job_type"],
                   "params": arguments["payload"]}
        parent = tasks_client.queue_path(PROJECT, LOCATION, QUEUE)
        task = {
            "http_request": {
                "http_method": tasks_v2.HttpMethod.POST,
                "url": WORKER_URL,
                "headers": {"Content-Type": "application/json"},
                "body": json.dumps(payload).encode()
            }
        }
        tasks_client.create_task(request={"parent": parent, "task": task})
        db.collection("job_status").document(job_id).set({
            "job_id": job_id, "job_type": arguments["job_type"],
            "status": "QUEUED", "created_at": datetime.utcnow().isoformat()
        })
        return [types.TextContent(type="text",
            text="Job enqueued. ID: " + job_id)]

elif name == "check_job_status":
        doc = db.collection("job_status").document(arguments["job_id"]).get()
        if doc.exists:
            return [types.TextContent(type="text", text=str(doc.to_dict()))]
        return [types.TextContent(type="text", text="Job not found")]

async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

It gives the following output,

# Agent query: "Generate bulk invoices for all Delhi orders in March"
Tool: enqueue_job({job_type: "bulk_invoice", payload: {city: "Delhi", month: "2026-03"}})

Job enqueued. ID: f3a9c2d1-8b44-4e71-a901-cc12f5e8ab90

# Agent query: "Check status of job f3a9c2d1"
Tool: check_job_status({job_id: "f3a9c2d1-8b44-4e71-a901-cc12f5e8ab90"})

{"job_id": "f3a9c2d1-8b44-4e71-a901-cc12f5e8ab90", "job_type": "bulk_invoice",
 "status": "COMPLETED", "created_at": "2026-04-11T04:10:00",
 "completed_at": "2026-04-11T04:12:35", "records_processed": 1842}

For production use, set Cloud Tasks retry configuration with max attempts and backoff to handle transient worker failures. Use Cloud Tasks dispatch rate limits to prevent overwhelming downstream services. Store job results and errors in Firestore so agents can report outcomes to users. For sensitive payloads, use OIDC authentication on the HTTP target so only the Cloud Tasks service account can invoke the worker endpoint.

Send your comments, suggestions or queries regarding this site to [email protected].