-
Notifications
You must be signed in to change notification settings - Fork 22
Expand file tree
/
Copy pathsample_mcp.py
More file actions
75 lines (61 loc) · 2 KB
/
Copy pathsample_mcp.py
File metadata and controls
75 lines (61 loc) · 2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import json
from pathlib import Path
from fastmcp.server import FastMCP
from pydantic import BaseModel
RECORDS = json.loads(Path(__file__).with_name("records.json").read_text())
LOOKUP = {r["id"]: r for r in RECORDS}
class SearchResult(BaseModel):
id: str
title: str
text: str
class SearchResultPage(BaseModel):
results: list[SearchResult]
class FetchResult(BaseModel):
id: str
title: str
text: str
url: str | None = None
metadata: dict[str, str] | None = None
def create_server():
mcp = FastMCP(name="Cupcake MCP", instructions="Search cupcake orders")
@mcp.tool()
async def search(query: str) -> SearchResultPage:
"""
Search for cupcake orders – keyword match.
Returns a SearchResultPage containing a list of SearchResult items.
"""
toks = query.lower().split()
results: list[SearchResult] = []
for r in RECORDS:
hay = " ".join(
[
r.get("title", ""),
r.get("text", ""),
" ".join(r.get("metadata", {}).values()),
]
).lower()
if any(t in hay for t in toks):
results.append(
SearchResult(id=r["id"], title=r.get("title", ""), text=r.get("text", ""))
)
# Return the Pydantic model (FastMCP will serialise it for us)
return SearchResultPage(results=results)
@mcp.tool()
async def fetch(id: str) -> FetchResult:
"""
Fetch a cupcake order by ID.
Returns a FetchResult model.
"""
if id not in LOOKUP:
raise ValueError("unknown id")
r = LOOKUP[id]
return FetchResult(
id=r["id"],
title=r.get("title", ""),
text=r.get("text", ""),
url=r.get("url"),
metadata=r.get("metadata"),
)
return mcp
if __name__ == "__main__":
create_server().run(transport="sse", host="127.0.0.1", port=8000)