skill.py

   1 from fastmcp import FastMCP
   2 from matrix_shop_bot.db import Data
   3 import os
   4 import uvicorn
   5
   6 # CORS
   7 from starlette.middleware import Middleware
   8 from starlette.middleware.cors import CORSMiddleware
   9
  10 # HTTP
  11 from starlette.applications import Starlette
  12 from starlette.routing import Mount
  13
  14
  15 MY_IP=os.environ.get('MY_IP')
  16 mcp = FastMCP("Shop Bot")
  17
  18 @mcp.tool
  19 def add_item(name: str) -> str:
  20     return '\n'.join(Data().add_item(name))
  21
  22
  23 # Configure CORS for browser-based clients
  24 middleware = [
  25     Middleware(
  26         CORSMiddleware,
  27         allow_origins=["*"],  # Allow all origins; use specific origins for security
  28         allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
  29         allow_headers=[
  30             "mcp-protocol-version",
  31             "mcp-session-id",
  32             "Authorization",
  33             "Content-Type",
  34         ],
  35         expose_headers=["mcp-session-id"],
  36     )
  37 ]
  38
  39 # if __name__ == "__main__":
  40
  41
  42
  43 #@mcp.tool
  44 #def buy(items:str) -> 'list[str]':
  45 #    return Data().purchase_items(items)
  46 #
  47 #@mcp.tool
  48 #def list_bought(store:str) -> 'list[str]':
  49 #    return Data().list_purchases(store)
  50 #
  51 #@mcp.tool
  52 #def clear(store_name: str, clear_all:bool=False) -> 'list[str]':
  53 #    return Data().clear_list(store_name, clear_all)
  54 #
  55 #
  56 #@mcp.tool
  57 #def shop_at(store_name:str) -> 'list[str]':
  58 #    '''Select store to shop at'''
  59 #    return Data().get_shopping_list(store_name)
  60 #
  61 #
  62 #@mcp.tool
  63 #def list() -> 'list[str]':
  64 #    '''Use previous store'''
  65 #    return Data().get_last_shopping_list()
  66 #
  67 #
  68 #@mcp.tool
  69 #def list_stores() -> 'list[str]':
  70 #    return Data().get_stores()
  71 #
  72 #
  73 #@mcp.tool
  74 #def list_purchases(store: str) -> 'list[str]':
  75 #    return Data().list_purchases(store)
  76
  77 if __name__ == "__main__":
  78 #   mcp.run(
  79 #           transport="http",
  80 #           host=MY_IP,
  81 #           stateless_http=True,
  82 #           port=9000)
  83
  84
  85        # Create the ASGI app
  86     mcp_app = mcp.http_app(
  87             path="/mcp",
  88             # middleware=middleware
  89             )
  90
  91     # Create a Starlette app and mount the MCP server
  92     app = Starlette(
  93         routes=[
  94             Mount("/mcp-server", app=mcp_app),
  95             # Add other routes as needed
  96         ],
  97         lifespan=mcp_app.lifespan,
  98     )
  99
 100     # http_app = mcp.http_app(transport="streamable-http",path="/",middleware=custom_middleware)
 101     http_app = mcp.http_app(transport="streamable-http",path="/",middleware=middleware)
 102     # http_app = mcp.http_app(transport="streamable-http",path="/")
 103     uvicorn.run(http_app, host=MY_IP, port=9000)
 104