1717This metadata helps clients better understand and work with the resource content.
1818"""
1919
20- from mcp .server .fastmcp import FastMCP
21- from mcp .types import TextResourceContents , BlobResourceContents
22- from pydantic import AnyUrl
2320import base64
2421
22+ from pydantic import AnyUrl
23+
24+ from mcp .server .fastmcp import FastMCP
25+ from mcp .types import BlobResourceContents , TextResourceContents
26+
2527mcp = FastMCP (name = "Direct ResourceContents Example" )
2628
2729
@@ -40,7 +42,7 @@ def get_report() -> TextResourceContents:
4042 "version" : "1.2.0" ,
4143 "tags" : ["monthly" , "finance" , "q1-2024" ],
4244 "confidentiality" : "internal" ,
43- }
45+ },
4446 )
4547
4648
@@ -50,7 +52,7 @@ def get_logo() -> BlobResourceContents:
5052 """Return a logo image with metadata about dimensions and format."""
5153 # In a real app, you might read this from a file
5254 image_bytes = b"\x89 PNG\r \n \x1a \n ..." # PNG header
53-
55+
5456 return BlobResourceContents (
5557 uri = AnyUrl ("image://logo" ),
5658 blob = base64 .b64encode (image_bytes ).decode (),
@@ -64,7 +66,7 @@ def get_logo() -> BlobResourceContents:
6466 "hasAlpha" : True ,
6567 "fileSize" : 24576 ,
6668 "lastModified" : "2024-01-10T08:00:00Z" ,
67- }
69+ },
6870 )
6971
7072
@@ -73,13 +75,14 @@ def get_logo() -> BlobResourceContents:
7375async def get_metrics (metric_type : str ) -> TextResourceContents :
7476 """Return metrics data with metadata about collection time and source."""
7577 import datetime
76-
78+
7779 # Simulate collecting metrics
7880 metrics = {"cpu" : 45.2 , "memory" : 78.5 , "disk" : 62.1 }
7981 timestamp = datetime .datetime .now (datetime .UTC ).isoformat ()
80-
82+
8183 if metric_type == "json" :
8284 import json
85+
8386 return TextResourceContents (
8487 uri = AnyUrl (f"data://metrics/{ metric_type } " ),
8588 text = json .dumps (metrics , indent = 2 ),
@@ -90,7 +93,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
9093 "interval" : "5s" ,
9194 "aggregation" : "average" ,
9295 "host" : "prod-server-01" ,
93- }
96+ },
9497 )
9598 elif metric_type == "csv" :
9699 csv_text = "metric,value\n " + "\n " .join (f"{ k } ,{ v } " for k , v in metrics .items ())
@@ -102,7 +105,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
102105 "timestamp" : timestamp ,
103106 "columns" : ["metric" , "value" ],
104107 "row_count" : len (metrics ),
105- }
108+ },
106109 )
107110 else :
108111 text = "\n " .join (f"{ k .upper ()} : { v } %" for k , v in metrics .items ())
@@ -113,7 +116,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
113116 _meta = {
114117 "timestamp" : timestamp ,
115118 "format" : "human-readable" ,
116- }
119+ },
117120 )
118121
119122
@@ -122,7 +125,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
122125def get_config () -> TextResourceContents :
123126 """Return application config with version and environment metadata."""
124127 import json
125-
128+
126129 config = {
127130 "version" : "1.0.0" ,
128131 "features" : {
@@ -133,9 +136,9 @@ def get_config() -> TextResourceContents:
133136 "limits" : {
134137 "max_file_size" : 10485760 , # 10MB
135138 "max_connections" : 100 ,
136- }
139+ },
137140 }
138-
141+
139142 return TextResourceContents (
140143 uri = AnyUrl ("config://app" ),
141144 text = json .dumps (config , indent = 2 ),
@@ -146,7 +149,7 @@ def get_config() -> TextResourceContents:
146149 "environment" : "production" ,
147150 "schema" : "https://example.com/schemas/config/v1.0" ,
148151 "editable" : False ,
149- }
152+ },
150153 )
151154
152155
@@ -156,7 +159,7 @@ async def get_users() -> TextResourceContents:
156159 """Return query results with execution time and row count."""
157160 import json
158161 import time
159-
162+
160163 # Simulate database query
161164 start_time = time .time ()
162165 users = [
@@ -165,7 +168,7 @@ async def get_users() -> TextResourceContents:
165168 {"id" : 3 , "name" : "Charlie" , "role" : "user" },
166169 ]
167170 execution_time = time .time () - start_time
168-
171+
169172 return TextResourceContents (
170173 uri = AnyUrl ("db://query/users" ),
171174 text = json .dumps (users , indent = 2 ),
@@ -177,10 +180,10 @@ async def get_users() -> TextResourceContents:
177180 "database" : "main" ,
178181 "cached" : False ,
179182 "timestamp" : "2024-01-15T16:00:00Z" ,
180- }
183+ },
181184 )
182185
183186
184187if __name__ == "__main__" :
185188 # Run with: python resource_contents_direct.py
186- mcp .run ()
189+ mcp .run ()
0 commit comments