跳转到主要内容
资源是 Client/LLM 可以拉取作为上下文的只读数据。与工具（动作）不同，资源旨在提供无副作用的信息。Client 也可以订阅资源 URI 并接收变更通知。 资源在模型上下文协议 (MCP) 中的流转方式如下：
  1. Client 通过 resources/list 发现资源（每个资源包含 uri、可选的 name、可选的 mimeType 等）。
  2. Client 通过携带 uri 的 resources/read 读取资源。
  3. 服务器执行你的资源处理器。
  4. 服务器返回符合 MCP 规范、包含 contents 的 ReadResourceResult。
使用 @resource(...) 定义资源，并通过 server.collect(...)（或在 with server.binding(): ... 代码块内）注册它们。

​
装饰器签名

  • resource(...): resource(uri: str, *, name=None, description=None, mime_type=None)
你的处理函数会按如下方式被调用：
  • fn() → 返回 str（文本）或 bytes（二进制）。（也支持使用 async def。）

​
基础资源

from dedalus_mcp import MCPServer, resource

@resource("resource://config/app", name="config", description="Application config", mime_type="application/json")
def app_config() -> str:
    return '{"debug": true, "version": "1.2.0"}'

server = MCPServer("config-server")
server.collect(app_config)
装饰器定义资源 URI，collect() 负责注册它。Client 使用 resources/list 列出资源，并通过 resources/read 读取它们。

​
文本与二进制

对于文本数据，返回 str： 
from dedalus_mcp import resource

@resource("resource://docs/readme", mime_type="text/markdown")
def readme() -> str:
    with open("README.md", "r", encoding="utf-8") as f:
        return f.read()
对于二进制数据返回 bytes 类型（Dedalus 会在 MCP 响应中将其编码为 base64）： 
from dedalus_mcp import resource

@resource("resource://assets/logo", mime_type="image/png")
def logo() -> bytes:
    with open("logo.png", "rb") as f:
        return f.read()

​
异步资源

import anyio
from dedalus_mcp import resource

@resource("resource://api/users", description="活跃用户", mime_type="application/json")
async def users() -> str:
    await anyio.sleep(0.1)  # 模拟 I/O 操作
    return '{"users": ["ada", "grace"]}'
对于 I/O，优先使用 async def。（与工具类似，同步处理函数会在当前执行流中直接运行。）

​
MIME 类型

import json
from dedalus_mcp import resource

@resource("resource://data/report", mime_type="application/json")
def report() -> str:
    return json.dumps({"users": 100, "active": 42})
通用默认值：
  • text/plain（当返回 str 且未提供 mime_type 时的默认值）
  • application/octet-stream（当返回 bytes 且未提供 mime_type 时的默认值）

​
订阅

Client 可以通过 resources/subscribe 订阅资源变更，并通过 resources/unsubscribe 取消订阅。 当底层数据发生变化时，请通知订阅方： 
from dedalus_mcp import MCPServer

server = MCPServer("live-data")
server.collect(users)

# 当此 URI 对应的数据发生变化时:
await server.notify_resource_updated("resource://api/users")
已订阅的 Client 将会收到 notifications/resources/updated。

​
装饰器选项

from dedalus_mcp import resource

@resource(
    "resource://config/app",
    name="app-config",           # 人类友好的名称(在 resources/list 中显示)
    description="App settings",  # Shown to clients
    mime_type="application/json",
)
def config() -> str:
    return '{"debug": false}'

​
错误处理

如果你的资源处理器抛出异常，Dedalus 会返回一个文本回退资源：
  • mimeType="text/plain"
  • text="Resource error: <exception message>"
如果某个 URI 未被注册，resources/read 会返回一个空的 contents 列表。

​
测试

像测试普通函数一样测试资源处理器： 
import json

def test_config_resource():
    data = json.loads(app_config())
    assert data["version"] == "1.2.0"
通过服务器的应用程序编程接口 API 进行集成式测试（对应 resources/read）： 
import pytest
from dedalus_mcp import MCPServer

@pytest.mark.asyncio
async def test_resource_read():
    server = MCPServer("test")
    server.collect(app_config)

    result = await server.invoke_resource("resource://config/app")
    assert result.contents[0].text == '{"debug": true, "version": "1.2.0"}'

​
资源模板

使用 @resource_template(...) 通过 resources/templates/list 暴露 URI 模式（便于 Client 发现和自动补全）。Dedalus 模型上下文协议 (MCP) 当前只会为列表操作注册模板，不会自动将 resources/read 路由到某个模板函数——你仍然需要使用 @resource(...) 注册具体的资源 URI。 
from dedalus_mcp import resource_template

@resource_template(
    "user-profile",
    uri_template="resource://users/{user_id}",
    description="User profile by ID",
)
async def user_profile(user_id: str) -> str:
    user = await fetch_user(user_id)
    return json.dumps(user)