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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Cloud Scheduler

MCP Server with Cloud Scheduler

Author: Venkata Sudhakar

Cloud Scheduler is a fully managed cron job service that runs on a configurable schedule and triggers HTTP, Pub/Sub, or App Engine targets. Exposing Cloud Scheduler through an MCP server gives AI agents the ability to manage scheduled jobs dynamically - creating new schedules, pausing existing ones, and triggering immediate runs on demand.

ShopMax India uses Cloud Scheduler for daily report generation, weekly supplier sync, and hourly price feed updates. The MCP server below lets the operations agent list active jobs, create new schedules, and force-run a job immediately when the business requires an off-cycle execution.

The below example shows an MCP server that manages Cloud Scheduler jobs as MCP tools, enabling agents to control scheduled operations programmatically.

# Install: pip install mcp google-cloud-scheduler
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from google.cloud import scheduler_v1
from google.protobuf import field_mask_pb2

PROJECT = "shopmax-india"
LOCATION = "asia-south1"
PARENT = "projects/{}/locations/{}".format(PROJECT, LOCATION)

client = scheduler_v1.CloudSchedulerClient()
app = Server("scheduler-mcp")

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="list_jobs",
            description="List all Cloud Scheduler jobs",
            inputSchema={"type": "object", "properties": {}}
        ),
        types.Tool(
            name="create_job",
            description="Create a new Cloud Scheduler job",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "Job name e.g. daily-report"},
                    "schedule": {"type": "string", "description": "Cron expression e.g. 0 6 * * *"},
                    "url": {"type": "string", "description": "HTTP endpoint to trigger"},
                    "description": {"type": "string"}
                },
                "required": ["name", "schedule", "url"]
            }
        ),
        types.Tool(
            name="run_job_now",
            description="Force run a Cloud Scheduler job immediately",
            inputSchema={
                "type": "object",
                "properties": {
                    "job_name": {"type": "string", "description": "Full job name or short name"}
                },
                "required": ["job_name"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "list_jobs":
        jobs = client.list_jobs(request={"parent": PARENT})
        result = []
        for job in jobs:
            result.append({"name": job.name.split("/")[-1],
                           "schedule": job.schedule,
                           "state": scheduler_v1.Job.State(job.state).name})
        return [types.TextContent(type="text", text=str(result))]

elif name == "create_job":
        job = scheduler_v1.Job(
            name="{}/jobs/{}".format(PARENT, arguments["name"]),
            description=arguments.get("description", ""),
            schedule=arguments["schedule"],
            time_zone="Asia/Kolkata",
            http_target=scheduler_v1.HttpTarget(
                uri=arguments["url"],
                http_method=scheduler_v1.HttpMethod.POST
            )
        )
        client.create_job(request={"parent": PARENT, "job": job})
        return [types.TextContent(type="text",
            text="Job created: " + arguments["name"])]

elif name == "run_job_now":
        job_name = arguments["job_name"]
        if "/" not in job_name:
            job_name = "{}/jobs/{}".format(PARENT, job_name)
        client.run_job(request={"name": job_name})
        return [types.TextContent(type="text", text="Job triggered successfully")]

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: "List all scheduled jobs"
Tool: list_jobs({})

[{"name": "daily-sales-report", "schedule": "0 6 * * *", "state": "ENABLED"},
 {"name": "hourly-price-feed", "schedule": "0 * * * *", "state": "ENABLED"},
 {"name": "weekly-supplier-sync", "schedule": "0 2 * * 1", "state": "ENABLED"}]

# Agent query: "Create a new job to run stock audit every night at 11 PM"
Tool: create_job({name: "nightly-stock-audit", schedule: "0 23 * * *",
                  url: "https://ops.shopmax.in/audit/stock"})

Job created: nightly-stock-audit

# Agent query: "Run the daily-sales-report job right now"
Tool: run_job_now({job_name: "daily-sales-report"})

Job triggered successfully

Set Cloud Scheduler jobs to use OIDC tokens when targeting Cloud Run or internal services so only authenticated callers can invoke the endpoints. Use the Asia/Kolkata timezone for ShopMax jobs to ensure schedules align with Indian business hours. Avoid creating jobs with sub-minute intervals as Cloud Scheduler has a one-minute minimum frequency; use Cloud Tasks or Pub/Sub for higher-frequency work.

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