|
| 1 | +import json |
| 2 | +import os |
| 3 | + |
| 4 | +from mcp.server.fastmcp import FastMCP |
| 5 | +from requests_oauthlib import OAuth1Session |
| 6 | + |
| 7 | + |
| 8 | +# Initialize FastMCP server |
| 9 | +mcp = FastMCP( |
| 10 | + "DevHub CMS MCP", |
| 11 | + description="Integration with DevHub CMS to manage content") |
| 12 | + |
| 13 | +devhub_api = OAuth1Session( |
| 14 | + os.environ['DEVHUB_API_KEY'], |
| 15 | + client_secret=os.environ['DEVHUB_API_SECRET']) |
| 16 | +base_url = '{}/api/v2/'.format(os.environ['DEVHUB_BASE_URL']) |
| 17 | + |
| 18 | + |
| 19 | +@mcp.tool() |
| 20 | +def get_hours_of_operation(location_id: int, hours_type: str = 'primary') -> list: |
| 21 | + """Get the hours of operation for a DevHub location |
| 22 | +
|
| 23 | + Returns a list of items representing days of the week |
| 24 | +
|
| 25 | + Except for the special case formatting, this object is a list of 7 items which represent each day. |
| 26 | +
|
| 27 | + Each day can can have one-four time ranges. For example, two time ranges denotes a "lunch-break". No time ranges denotes closed. |
| 28 | +
|
| 29 | + Examples: |
| 30 | + 9am-5pm [["09:00:00", "17:00:00"]] |
| 31 | + 9am-12pm and 1pm-5pm [["09:00:00", "12:00:00"], ["13:00:00", "17:00:00"]] |
| 32 | + Closed - an empty list [] |
| 33 | +
|
| 34 | + Args: |
| 35 | + location_id: DevHub Location ID |
| 36 | + hours_type: Defaults to 'primary' unless the user specifies a different type |
| 37 | + """ |
| 38 | + r = devhub_api.get('{}locations/{}'.format(base_url, location_id)) |
| 39 | + content = json.loads(r.content) |
| 40 | + return content['hours_by_type'].get(hours_type, []) |
| 41 | + |
| 42 | + |
| 43 | +@mcp.tool() |
| 44 | +def update_hours(location_id: int, new_hours: list, hours_type: str = 'primary') -> str: |
| 45 | + """Update the hours of operation for a DevHub location |
| 46 | +
|
| 47 | + Send a list of items representing days of the week |
| 48 | +
|
| 49 | + Except for the special case formatting, this object is a list of 7 items which represent each day. |
| 50 | +
|
| 51 | + Each day can can have one-four time ranges. For example, two time ranges denotes a "lunch-break". No time ranges denotes closed. |
| 52 | +
|
| 53 | + Examples: |
| 54 | + 9am-5pm [["09:00:00", "17:00:00"]] |
| 55 | + 9am-12pm and 1pm-5pm [["09:00:00", "12:00:00"], ["13:00:00", "17:00:00"]] |
| 56 | + Closed - an empty list [] |
| 57 | +
|
| 58 | + Args: |
| 59 | + location_id: DevHub Location ID |
| 60 | + new_hours: Structured format of the new hours |
| 61 | + hours_type: Defaults to 'primary' unless the user specifies a different type |
| 62 | + """ |
| 63 | + r = devhub_api.put( |
| 64 | + '{}locations/{}/'.format(base_url, location_id), |
| 65 | + json={ |
| 66 | + 'hours': [ |
| 67 | + { |
| 68 | + 'type': hours_type, |
| 69 | + 'hours': new_hours, |
| 70 | + } |
| 71 | + ] |
| 72 | + }, |
| 73 | + ) |
| 74 | + content = json.loads(r.content) |
| 75 | + return 'Updated successfully' |
| 76 | + |
| 77 | + |
| 78 | +@mcp.tool() |
| 79 | +def upload_image(base64_image_content: str, filename: str) -> str: |
| 80 | + """Upload an image to the DevHub media gallery |
| 81 | +
|
| 82 | + Supports webp, jpeg and png images |
| 83 | +
|
| 84 | + Args: |
| 85 | + base64_image_content: Base 64 encoded content of the image file |
| 86 | + filename: Filename including the extension |
| 87 | + """ |
| 88 | + payload = { |
| 89 | + 'type': 'image', |
| 90 | + 'upload': { |
| 91 | + 'file': base64_image_content, |
| 92 | + 'filename': filename, |
| 93 | + } |
| 94 | + } |
| 95 | + r = devhub_api.post( |
| 96 | + '{}images/'.format(base_url), |
| 97 | + json=payload, |
| 98 | + ) |
| 99 | + image = r.json() |
| 100 | + return f""" |
| 101 | +Image ID: {image['id']} |
| 102 | +Image Path (for use in HTML src attributes): {image['absolute_path']} |
| 103 | +""" |
| 104 | + |
| 105 | + |
| 106 | +@mcp.tool() |
| 107 | +def get_blog_post(post_id: int) -> str: |
| 108 | + """Get a single blog post |
| 109 | +
|
| 110 | + Args: |
| 111 | + post_id: Blog post id |
| 112 | + """ |
| 113 | + r = devhub_api.get('{}posts/{}/'.format(base_url, post_id)) |
| 114 | + post = r.json() |
| 115 | + return f""" |
| 116 | +Post ID: {post['id']} |
| 117 | +Title: {post['title']} |
| 118 | +Date: {post['date']} |
| 119 | +
|
| 120 | +Content (HTML): |
| 121 | +{post['content']} |
| 122 | +""" |
| 123 | + |
| 124 | + |
| 125 | +@mcp.tool() |
| 126 | +def create_blog_post(site_id: int, title: str, content: str) -> str: |
| 127 | + """Create a new blog post |
| 128 | +
|
| 129 | + Args: |
| 130 | + site_id: Website ID where the post will be published. Prompt the user for this ID. |
| 131 | + title: Blog post title |
| 132 | + content: HTML content of blog post. Should not include a <h1> tag, only h2+ |
| 133 | + """ |
| 134 | + payload = {} |
| 135 | + payload['site_id'] = site_id |
| 136 | + payload['content'] = content |
| 137 | + payload['title'] = title |
| 138 | + r = devhub_api.post( |
| 139 | + '{}posts/'.format(base_url), |
| 140 | + json=payload, |
| 141 | + ) |
| 142 | + post = r.json() |
| 143 | + return f""" |
| 144 | +Post ID: {post['id']} |
| 145 | +Title: {post['title']} |
| 146 | +Date: {post['date']} |
| 147 | +
|
| 148 | +Content (HTML): |
| 149 | +{post['content']} |
| 150 | +""" |
| 151 | + |
| 152 | + |
| 153 | +@mcp.tool() |
| 154 | +def update_blog_post(post_id: int, title: str = None, content: str = None) -> str: |
| 155 | + """Update a single blog post |
| 156 | +
|
| 157 | + Args: |
| 158 | + post_id: Blog post ID |
| 159 | + title: Blog post title |
| 160 | + content: HTML content of blog post. Should not include a <h1> tag, only h2+ |
| 161 | + """ |
| 162 | + payload = {} |
| 163 | + if content: |
| 164 | + payload['content'] = content |
| 165 | + if title: |
| 166 | + payload['title'] = title |
| 167 | + r = devhub_api.put( |
| 168 | + '{}posts/{}/'.format(base_url, post_id), |
| 169 | + json=payload, |
| 170 | + ) |
| 171 | + post = r.json() |
| 172 | + return f""" |
| 173 | +Post ID: {post['id']} |
| 174 | +Title: {post['title']} |
| 175 | +Date: {post['date']} |
| 176 | +
|
| 177 | +Content (HTML): |
| 178 | +{post['content']} |
| 179 | +""" |
| 180 | + |
| 181 | + |
| 182 | +@mcp.tool() |
| 183 | +def get_nearest_location(business_id: int, latitude: float, longitude: float) -> str: |
| 184 | + """Get the nearest DevHub location |
| 185 | +
|
| 186 | + Args: |
| 187 | + business_id: DevHub Business ID associated with the location. Prompt the user for this ID |
| 188 | + latitude: Latitude of the location |
| 189 | + longitude: Longitude of the location |
| 190 | + """ |
| 191 | + r = devhub_api.get('{}locations/'.format(base_url), params={ |
| 192 | + 'business_id': business_id, |
| 193 | + 'near_lat': latitude, |
| 194 | + 'near_lon': longitude, |
| 195 | + }) |
| 196 | + objects = json.loads(r.content)['objects'] |
| 197 | + if objects: |
| 198 | + return f""" |
| 199 | +Location ID: {objects[0]['id']} |
| 200 | +Location name: {objects[0]['location_name']} |
| 201 | +Location url: {objects[0]['location_url']} |
| 202 | +Street: {objects[0]['street']} |
| 203 | +City: {objects[0]['city']} |
| 204 | +State: {objects[0]['state']} |
| 205 | +Country: {objects[0]['country']} |
| 206 | +""" |
| 207 | + |
| 208 | + |
| 209 | +def main(): |
| 210 | + """Run the MCP server""" |
| 211 | + mcp.run(transport='stdio') |
| 212 | + |
| 213 | +if __name__ == "__main__": |
| 214 | + main() |
0 commit comments