Task Scheduler MCP Server (Python)

RabbitMQ + Celery + Redis/SQLite 를 사용하는 태스크 스케줄러 MCP 서버 예제입니다.

주요 기능

• 즉시 비동기 실행: 시간이 오래 걸리는 작업을 Celery 로 백그라운드 실행, MCP 응답은 즉시 반환.

• 예약 실행 (Scheduling): ISO 8601 시간(UTC 기준)으로 실행 시점 예약.

• 작업 관리: 작업 상태 조회, 취소.

• 결과 알림: 작업 완료 시 KakaoTalk(추후 구현) 또는 로그 기반 Notifier 로 결과 전송.

기술 스택

• Broker: RabbitMQ

• Backend: Redis (선호) 또는 SQLite (SQLAlchemy result backend)

• Worker: Celery

• Interface: MCP Python SDK + FastAPI(HTTP 서버)

설치

pip install -r requirements.txt

RabbitMQ, Redis 는 시스템 레벨에서 미리 띄워져 있어야 합니다.

예시:

# RabbitMQ (docker 예시)
docker run -d --name rabbitmq -p 5672:5672 rabbitmq:3-management

# Redis (선택)
docker run -d --name redis -p 6379:6379 redis:7

환경 변수

• BROKER_URL: Celery broker URL (기본값 amqp://guest:guest@localhost:5672//)

• REDIS_URL: Celery result backend 로 사용할 Redis URL (예: redis://localhost:6379/0)

• SQLITE_URL: SQLite 기반 result backend URL (기본값 db+sqlite:///./celery_results.sqlite3)

• HOST, PORT: MCP HTTP 서버 바인딩 정보 (기본 0.0.0.0:8000)

실행 방법

1. Celery 워커 실행

celery -A celery_app.celery_app worker --loglevel=info

2. MCP HTTP 서버 실행

python mcp_server.py

이제 http://localhost:8000/health 에서 헬스 체크가 가능합니다.

MCP 도구 개요

mcp_server.py 내 FastMCP("task-scheduler-mcp") 기반으로 다음 도구가 정의되어 있습니다.

• enqueue_task(payload, notify_target=None)

  • 설명: 즉시 비동기 실행. Celery 로 바로 태스크를 던지고 MCP 에는 task_id 만 반환.

  • 파라미터:

    • payload: 실제 작업에 필요한 데이터(임의의 JSON).

    • notify_target: 작업 완료 시 알림을 받을 대상 식별자(예: 카카오톡 user id).

• schedule_task(run_at, payload, notify_target=None)

  • 설명: 예약 실행. run_at 시각(ISO 8601, UTC 기준)에 Celery ETA 기능으로 실행.

  • 파라미터:

    • run_at: 예) "2026-01-09T12:00:00Z".

    • payload, notify_target: 위와 동일.

• get_task_status(task_id)

  • 설명: 특정 작업의 상태 및 결과 조회.

  • 반환: task_id, status, result 필드를 포함한 JSON.

• cancel_task(task_id)

  • 설명: 대기 중/실행 중인 작업을 취소(revoke) 시도.

MCP 클라이언트(예: Cursor 또는 OpenAI IDE 플러그인)를 사용할 때는 해당 도구들을 노출하는 MCP 서버로 이 HTTP 엔드포인트를 등록하면 됩니다.

KakaoTalk 연동 포인트

notification.py 에서 다음과 같이 인터페이스를 정의합니다.

• Notifier (추상 베이스 클래스)

• LogNotifier (기본 구현: 로그 출력)

• KakaoTalkNotifier (골격/모킹 구현: 실제 REST API 연동은 TODO)

실제 카카오톡 연동 시:

1. KakaoTalk REST API 토큰/앱 키를 환경 변수나 설정 파일로 받고,

2. KakaoTalkNotifier.notify() 내에서 카카오 메시지 전송 API를 호출하도록 구현하면 됩니다.

현재 기본 동작은 LogNotifier 로 설정되어 있어, 작업 완료 시 로그에 결과가 기록됩니다.