mcp-tool.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("Shopping List")
  17
  18 @mcp.tool
  19 def add_item(name: str) -> str:
  20     return str(Data().add_item(name))
  21
  22 @mcp.tool
  23 def buy(items:str) -> str:
  24     return '\n'.join(Data().purchase_items(items))
  25
  26 @mcp.tool
  27 def list_bought(store:str) -> str:
  28     return '\n'.join(Data().list_purchases(store))
  29
  30 #@mcp.tool
  31 #def clear(store_name: str, clear_all:bool=False) -> 'list[str]':
  32 #    return Data().clear_list(store_name, clear_all)
  33
  34
  35 @mcp.tool
  36 def shop_at(store_name:str) -> str:
  37     '''Select store to shop at'''
  38     return '\n'.join(Data().get_shopping_list(store_name))
  39
  40
  41 @mcp.tool
  42 def list() -> str:
  43     '''Use previous store'''
  44     return '\n'.join(Data().get_last_shopping_list())
  45
  46
  47 #@mcp.tool
  48 #def list_stores() -> 'list[str]':
  49 #    return Data().get_stores()
  50
  51
  52 #@mcp.tool
  53 #def list_purchases(store: str) -> 'list[str]':
  54 #    return Data().list_purchases(store)
  55
  56 # Configure CORS for browser-based clients
  57 middleware = [
  58     Middleware(
  59         CORSMiddleware,
  60         allow_origins=["*"],  # Allow all origins; use specific origins for security
  61         allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
  62         allow_headers=[
  63             "mcp-protocol-version",
  64             "mcp-session-id",
  65             "Authorization",
  66             "Content-Type",
  67         ],
  68         expose_headers=["mcp-session-id"],
  69     )
  70 ]
  71
  72
  73
  74 if __name__ == "__main__":
  75
  76        # Create the ASGI app
  77     mcp_app = mcp.http_app(
  78             path="/mcp",
  79             # middleware=middleware
  80             )
  81
  82     # Create a Starlette app and mount the MCP server
  83     app = Starlette(
  84         routes=[
  85             Mount("/mcp-server", app=mcp_app),
  86             # Add other routes as needed
  87         ],
  88         lifespan=mcp_app.lifespan,
  89     )
  90
  91     # http_app = mcp.http_app(transport="streamable-http",path="/",middleware=custom_middleware)
  92     http_app = mcp.http_app(transport="streamable-http",path="/",middleware=middleware)
  93     # http_app = mcp.http_app(transport="streamable-http",path="/")
  94     uvicorn.run(http_app, host=MY_IP, port=9000)
  95