We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/inspirit941/kakao-bot-mcp-server'
If you have feedback or need assistance with the MCP directory API, please join our Discord server
from typing import Sequence, Dict, Any
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
from mcp_kakao import toolhandler
from mcp_kakao.kauth import (
get_stored_credentials,
refresh_token,
TokenRefreshError,
get_authorization_url,
)
import json
import logging
import requests
from mcp_kakao.message import KakaoMessageService
from pydantic import ValidationError
from api.message import (
TextTemplate,
FeedTemplate,
ListTemplate,
LocationTemplate,
CalendarTemplate,
CommerceTemplate,
# Ensure all template models are imported
)
class BaseKakaoTemplateToolHandler(toolhandler.ToolHandler):
"""Base handler for Kakao template messages."""
def __init__(self, name: str, description: str):
super().__init__(name)
self._description = description
def get_tool_description(self) -> Tool:
# This method should be implemented by subclasses to provide
# the specific schema for each template type.
raise NotImplementedError("Subclasses must implement get_tool_description")
def _handle_response(self, response: requests.Response) -> Sequence[TextContent]:
"""Handles the response from the Kakao API."""
if response.status_code == 200:
return [
TextContent(
type="text",
text=json.dumps(response.json()),
)
]
else:
logging.error(
f"Failed to send message: {response.status_code}, {response.text}"
)
return [
TextContent(
type="text",
text=f"Failed to send message: {response.status_code}, {response.text}",
)
]
def run_tool(
self, args: Dict[str, Any]
) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
try:
email_address = args.get(toolhandler.EMAIL_ADDRESS_ARG)
if not email_address:
return [
TextContent(
type="text",
text=f"Missing required parameter: {toolhandler.EMAIL_ADDRESS_ARG}",
)
]
credentials = get_stored_credentials(email_address)
if credentials is None or credentials.access_token_expired:
# Attempt to refresh if credentials exist but are expired
# If no credentials exist, refresh_token will raise an error if no refresh token
try:
credentials = refresh_token(
credentials, email_address=email_address
)
except TokenRefreshError as e:
logging.error(f"Token refresh failed: {e}")
# If refresh fails (e.g., expired refresh token), provide login URL
auth_url = get_authorization_url(
email_address=email_address, state=""
) # Provide login URL
return [
TextContent(
type="text",
text=f"Your Kakao token needs to be refreshed. Please log in again using this URL: {auth_url}",
)
]
# If credentials were None initially and refresh_token didn't raise, something is wrong,
# but the subsequent use of credentials will likely raise an error caught below.
# Ensure credentials are valid after potential refresh
if credentials is None or not credentials.access_token:
auth_url = get_authorization_url(
email_address=email_address, state=""
) # Provide login URL if still no valid credentials
return [
TextContent(
type="text",
text=f"Could not retrieve or refresh Kakao credentials. Please log in again using this URL: {auth_url}",
)
]
message_service = KakaoMessageService(
email_address=email_address, credential=credentials
)
response = self._send_specific_template(message_service, args)
return self._handle_response(response)
except (
ValidationError,
ValueError,
RuntimeError,
) as e:
logging.error(f"Error processing message: {str(e)}")
return [
TextContent(type="text", text=f"Error processing message: {str(e)}")
]
except Exception as e:
logging.error(f"An unexpected error occurred: {str(e)}")
return [
TextContent(type="text", text=f"An unexpected error occurred: {str(e)}")
]
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
"""Abstract method for subclasses to send the specific template type."""
raise NotImplementedError("Subclasses must implement _send_specific_template")
class SendTextTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_text_template_to_me", "Sends a Kakao Talk text message to me."
)
def get_tool_description(self) -> Tool:
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": ["text", "link", toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
"text": {"type": "string", "maxLength": 200},
"link": {
"type": "object",
"properties": {
"web_url": {"type": "string", "format": "uri"},
"mobile_web_url": {"type": "string", "format": "uri"},
},
},
"button_title": {"type": "string"},
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
# The schema ensures args contains the necessary keys for TextTemplate
template = TextTemplate(**args)
return message_service.send_text_template(template)
class SendFeedTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_feed_template_to_me", "Sends a Kakao Talk feed message to me."
)
def get_tool_description(self) -> Tool:
# Extract the schema for 'feed' from the original oneOf block
feed_schema_properties = {
"content": {
"type": "object",
"required": [
"title",
"description",
"image_url",
"link",
],
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"image_url": {
"type": "string",
"format": "uri",
},
"image_width": {"type": "integer"},
"image_height": {"type": "integer"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
"item_content": {
"type": "object",
"properties": {
"profile_text": {"type": "string"},
"profile_image_url": {
"type": "string",
"format": "uri",
},
"title_image_url": {
"type": "string",
"format": "uri",
},
"title_image_text": {"type": "string"},
"title_image_category": {"type": "string"},
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"item": {"type": "string"},
"item_op": {"type": "string"},
},
},
},
"sum": {"type": "string"},
"sum_op": {"type": "string"},
},
},
"social": {
"type": "object",
"properties": {
"like_count": {"type": "integer"},
"comment_count": {"type": "integer"},
"shared_count": {"type": "integer"},
"view_count": {"type": "integer"},
"subscriber_count": {"type": "integer"},
},
},
"buttons": {
"type": "array",
"items": {
"type": "object",
"required": ["title", "link"],
"properties": {
"title": {"type": "string"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
}
feed_required_fields = ["content"] # Exclude object_type as it's implicit
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": feed_required_fields + [toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
**feed_schema_properties,
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
# The schema ensures args contains the necessary keys for FeedTemplate
# We need to manually add object_type="feed" for Pydantic validation if it's not in args
# However, since the schema matches the template model structure, Pydantic should handle it
template = FeedTemplate(**args)
return message_service.send_feed_template(template)
class SendListTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_list_template_to_me", "Sends a Kakao Talk list message to me."
)
def get_tool_description(self) -> Tool:
# Extract the schema for 'list' from the original oneOf block
list_schema_properties = {
"header_title": {"type": "string"},
"header_link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
"contents": {
"type": "array",
"items": {
"type": "object",
"required": [
"title",
"description",
"image_url",
"link",
],
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"image_url": {
"type": "string",
"format": "uri",
},
"image_width": {"type": "integer"},
"image_height": {"type": "integer"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
"buttons": {
"type": "array",
"items": {
"type": "object",
"required": ["title", "link"],
"properties": {
"title": {"type": "string"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
}
list_required_fields = ["header_title", "contents"] # Exclude object_type
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": list_required_fields + [toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
**list_schema_properties,
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
template = ListTemplate(**args)
return message_service.send_list_template(template)
class SendLocationTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_location_template_to_me", "Sends a Kakao Talk location message to me."
)
def get_tool_description(self) -> Tool:
# Extract the schema for 'location' from the original oneOf block
location_schema_properties = {
"content": {
"type": "object",
"required": [
"title",
"description",
"image_url",
"link",
],
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"image_url": {
"type": "string",
"format": "uri",
},
"image_width": {"type": "integer"},
"image_height": {"type": "integer"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
"buttons": {
"type": "array",
"items": {
"type": "object",
"required": ["title", "link"],
"properties": {
"title": {"type": "string"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
"address": {"type": "string"},
"address_title": {"type": "string"},
}
location_required_fields = ["content", "address"] # Exclude object_type
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": location_required_fields + [toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
**location_schema_properties,
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
template = LocationTemplate(**args)
return message_service.send_location_template(template)
class SendCalendarTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_calendar_template_to_me", "Sends a Kakao Talk calendar message to me."
)
def get_tool_description(self) -> Tool:
# Extract the schema for 'calendar' from the original oneOf block
calendar_schema_properties = {
"content": {
"type": "object",
"required": ["title", "description", "link"],
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"image_url": {
"type": "string",
"format": "uri",
},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
"buttons": {
"type": "array",
"items": {
"type": "object",
"required": ["title", "link"],
"properties": {
"title": {"type": "string"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
"id_type": {"type": "string", "enum": ["event"]},
"id": {"type": "string"},
}
calendar_required_fields = ["content", "id_type", "id"] # Exclude object_type
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": calendar_required_fields + [toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
**calendar_schema_properties,
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
template = CalendarTemplate(**args)
return message_service.send_calendar_template(template)
class SendCommerceTemplateToMeToolHandler(BaseKakaoTemplateToolHandler):
def __init__(self):
super().__init__(
"send_commerce_template_to_me", "Sends a Kakao Talk commerce message to me."
)
def get_tool_description(self) -> Tool:
# Extract the schema for 'commerce' from the original oneOf block
commerce_schema_properties = {
"content": {
"type": "object",
"required": ["title", "image_url", "link"],
"properties": {
"title": {"type": "string"},
"image_url": {
"type": "string",
"format": "uri",
},
"image_width": {"type": "integer"},
"image_height": {"type": "integer"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
"commerce": {
"type": "object",
"required": ["regular_price"],
"properties": {
"regular_price": {"type": "integer"},
"discount_price": {"type": "integer"},
"discount_rate": {
"type": "integer",
"minimum": 0,
"maximum": 100,
},
},
},
"buttons": {
"type": "array",
"items": {
"type": "object",
"required": ["title", "link"],
"properties": {
"title": {"type": "string"},
"link": {
"type": "object",
"properties": {
"web_url": {
"type": "string",
"format": "uri",
},
"mobile_web_url": {
"type": "string",
"format": "uri",
},
"android_execution_params": {"type": "string"},
"ios_execution_params": {"type": "string"},
},
},
},
},
},
}
commerce_required_fields = ["content", "commerce"] # Exclude object_type
return Tool(
name=self.name,
description=self._description,
inputSchema={
"type": "object",
"required": commerce_required_fields + [toolhandler.EMAIL_ADDRESS_ARG],
"properties": {
toolhandler.EMAIL_ADDRESS_ARG: self.get_email_address_arg_schema(),
**commerce_schema_properties,
},
},
)
def _send_specific_template(
self, message_service: KakaoMessageService, args: Dict[str, Any]
) -> requests.Response:
template = CommerceTemplate(**args)
return message_service.send_commerce_template(template)