forked from panz2018/fastapi_mcp_sse
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmatchmaking.py
More file actions
77 lines (61 loc) · 2.32 KB
/
Copy pathmatchmaking.py
File metadata and controls
77 lines (61 loc) · 2.32 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
76
77
from typing import Any, Dict
from mcp.server.fastmcp import FastMCP
import uuid
# Initialize FastMCP server
mcp = FastMCP("matchmaking")
# In-memory database for agents and matches
# For a production system, you would use a persistent database.
agents: Dict[str, Dict[str, Any]] = {}
matches: Dict[str, str] = {}
@mcp.tool()
async def announce_agent(profile: Dict[str, Any]) -> str:
"""
Announces an agent to the matchmaking service and returns a fingerprint.
Args:
profile: A dictionary containing the agent's profile, including skills,
orientations, preferences, and city.
"""
fingerprint = str(uuid.uuid4())
agents[fingerprint] = profile
return fingerprint
@mcp.tool()
async def get_potential_match(fingerprint: str) -> Dict[str, Any]:
"""
Gets a potential match for a given agent.
Args:
fingerprint: The fingerprint of the agent requesting a match.
Returns:
A dictionary containing the profile of a potential match, or an empty
dictionary if no match is found.
"""
if fingerprint not in agents:
return {"error": "Invalid fingerprint."}
for other_fingerprint, other_profile in agents.items():
if other_fingerprint == fingerprint:
continue
# Check if they are already matched
if matches.get(fingerprint) == other_fingerprint or \
matches.get(other_fingerprint) == fingerprint:
continue
# Simple matching logic: return the first available agent.
# A real implementation would have more complex matching logic.
return other_profile
return {}
@mcp.tool()
async def establish_match(fingerprint1: str, fingerprint2: str) -> str:
"""
Establishes a match between two agents, preventing them from being matched
with each other again.
Args:
fingerprint1: The fingerprint of the first agent.
fingerprint2: The fingerprint of the second agent.
Returns:
A confirmation message.
"""
if fingerprint1 not in agents or fingerprint2 not in agents:
return "Error: One or both fingerprints are invalid."
matches[fingerprint1] = fingerprint2
return f"Match established between {fingerprint1} and {fingerprint2}."
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport="sse")