1+ import asyncio
2+ from linkup import LinkupClient
3+ from mcp .server .models import InitializationOptions
4+ import mcp .types as types
5+ from mcp .server import NotificationOptions , Server
6+ import mcp .server .stdio
7+ from pydantic import AnyUrl
8+ import logging
9+
10+ import os
11+ # Initialize the LinkupClient
12+
13+ client = LinkupClient ()
14+
15+ server = Server ("mcp-search-linkup" )
16+ logger = logging .getLogger ("mcp-search-linkup" )
17+ logger .setLevel (logging .INFO )
18+
19+ ## Logging
20+ @server .set_logging_level ()
21+ async def set_logging_level (level : types .LoggingLevel ) -> types .EmptyResult :
22+ logger .setLevel (level .upper ())
23+ await server .request_context .session .send_log_message (
24+ level = "info" ,
25+ data = f"Log level set to { level } " ,
26+ logger = "mcp-search-linkup"
27+ )
28+ return types .EmptyResult ()
29+
30+ ## Resources
31+ @server .list_resources ()
32+ async def list_resources () -> list [types .Resource ]:
33+ return [
34+ types .Resource (
35+ uri = AnyUrl ("https://www.thebridgechronicle.com/media" ),
36+ name = "The Bridge Chronicle" ,
37+ mimeType = "text/html"
38+ )
39+ ]
40+
41+ @server .read_resource ()
42+ async def read_resource (uri : AnyUrl ) -> str :
43+ if str (uri ) == "https://www.thebridgechronicle.com/media" :
44+ page = client .content (url = "https://www.thebridgechronicle.com/news/capgemini-employees-walk-together-in-celebration-of-indias-independence" )
45+ return page .content
46+
47+
48+ raise ValueError ("Resource not found" )
49+
50+ ## Tools
51+ @server .list_tools ()
52+ async def handle_list_tools () -> list [types .Tool ]:
53+ """
54+ List available search tools.
55+ """
56+ return [
57+ types .Tool (
58+ name = "search-web" ,
59+ description = "Perform a web search query using Linkup. This tool is helpful for finding information on the web." ,
60+ inputSchema = {
61+ "type" : "object" ,
62+ "properties" : {
63+ "query" : {"type" : "string" },
64+ "depth" : {
65+ "type" : "string" ,
66+ "enum" : ["standard" , "deep" ],
67+ "default" : "standard"
68+ },
69+ "output_type" : {
70+ "type" : "string" ,
71+ "enum" : ["searchResults" , "sourcedAnswer" , "structured" ],
72+ "default" : "sourcedAnswer"
73+ }
74+ },
75+ "required" : ["query" ],
76+ },
77+ )
78+ ]
79+
80+ @server .call_tool ()
81+ async def handle_call_tool (
82+ name : str , arguments : dict | None
83+ ) -> list [types .TextContent | types .ImageContent | types .EmbeddedResource ]:
84+ """
85+ Handle search tool execution requests.
86+ """
87+ if name != "search-web" :
88+ raise ValueError (f"Unknown tool: { name } " )
89+
90+ if not arguments :
91+ raise ValueError ("Missing arguments" )
92+
93+ query = arguments .get ("query" )
94+ depth = arguments .get ("depth" , "standard" )
95+ output_type = arguments .get ("output_type" , "sourcedAnswer" )
96+
97+ if not query :
98+ raise ValueError ("Missing query" )
99+
100+ # Perform the search using LinkupClient
101+ search_response = client .search (
102+ query = query ,
103+ depth = depth ,
104+ output_type = output_type ,
105+ )
106+
107+ return [
108+ types .TextContent (
109+ type = "text" ,
110+ text = str (search_response ),
111+ )
112+ ]
113+
114+ async def main ():
115+ # Run the server using stdin/stdout streams
116+ async with mcp .server .stdio .stdio_server () as (read_stream , write_stream ):
117+ await server .run (
118+ read_stream ,
119+ write_stream ,
120+ InitializationOptions (
121+ server_name = "mcp-search-linkup" ,
122+ server_version = "0.1.0" ,
123+ capabilities = server .get_capabilities (
124+ notification_options = NotificationOptions (),
125+ experimental_capabilities = {},
126+ ),
127+ ),
128+ )
129+
130+ if __name__ == "__main__" :
131+ asyncio .run (main ())
0 commit comments