Turn crew-dcs callables into FastMCP v3 tools via a single generic factory.
A crew-dcs entry point is shaped (auth, <domain params>, *, <plumbing>, **context_kwargs).
build_tool_fn synthesizes an async wrapper whose public signature is
(instance: str, <domain params>, ctx) — plumbing stripped — that resolves the
named instance to a DomoAuth, invokes the target, and serializes the result.
build_tool_fn(
target: Callable[..., Awaitable[Any]],
auth_resolver: AuthResolver,
*,
name: str | None = None
) -> Callable[..., Awaitable[Any]]
Build an async MCP tool wrapper around a crew-dcs callable.
Source code in src/crew_dcs/mcp_server/factory.py
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135 | def build_tool_fn(
target: Callable[..., Awaitable[Any]],
auth_resolver: AuthResolver,
*,
name: str | None = None,
) -> Callable[..., Awaitable[Any]]:
"""Build an async MCP tool wrapper around a crew-dcs callable."""
domain = _domain_parameters(target)
accepts_ctx = "ctx" in inspect.signature(target).parameters
async def wrapper(instance: str, ctx: Context | None = None, **domain_kwargs):
verbose = bool(domain_kwargs.pop("verbose", False))
auth = await auth_resolver(instance, _extract_dc_token(ctx))
# Logger rides on the context (build_context preserves it). debug_api /
# log_level are passed as kwargs so they propagate through each method's
# build_context instead of being reset by per-method defaults.
route_context = RouteContext(
logger=MCPContextLogger(ctx) if ctx is not None else None,
)
if verbose:
domain_kwargs["debug_api"] = True
domain_kwargs["log_level"] = "DEBUG"
if accepts_ctx:
domain_kwargs["ctx"] = ctx
result = await target(auth=auth, context=route_context, **domain_kwargs)
return to_mcp_json(result)
# Synthesize the public signature: instance, <domain params>, ctx.
# Required params (no default) precede defaulted params to stay valid.
required = [p for p in domain if p.default is inspect.Parameter.empty]
defaulted = [p for p in domain if p.default is not inspect.Parameter.empty]
new_params = [
inspect.Parameter(
"instance", inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=str
),
*[p.replace(kind=inspect.Parameter.POSITIONAL_OR_KEYWORD) for p in required],
*[p.replace(kind=inspect.Parameter.POSITIONAL_OR_KEYWORD) for p in defaulted],
inspect.Parameter(
"verbose",
inspect.Parameter.POSITIONAL_OR_KEYWORD,
annotation=bool,
default=False,
),
inspect.Parameter(
"ctx",
inspect.Parameter.POSITIONAL_OR_KEYWORD,
annotation=Context | None,
default=None,
),
]
wrapper.__signature__ = inspect.Signature(new_params)
# FastMCP/pydantic build the schema from get_type_hints (i.e. __annotations__),
# so keep annotations in sync with the synthesized signature.
wrapper.__annotations__ = {
p.name: (p.annotation if p.annotation is not inspect.Parameter.empty else Any)
for p in new_params
}
wrapper.__name__ = name or getattr(target, "__name__", "tool")
wrapper.__doc__ = inspect.getdoc(target)
return wrapper
|
make_tool(
target: Callable[..., Awaitable[Any]],
auth_resolver: AuthResolver,
*,
name: str | None = None,
tags: set[str] | None = None,
annotations: Any | None = None
) -> Tool
Build and register a FastMCP tool from a crew-dcs callable.
Source code in src/crew_dcs/mcp_server/factory.py
138
139
140
141
142
143
144
145
146
147
148 | def make_tool(
target: Callable[..., Awaitable[Any]],
auth_resolver: AuthResolver,
*,
name: str | None = None,
tags: set[str] | None = None,
annotations: Any | None = None,
) -> Tool:
"""Build and register a FastMCP tool from a crew-dcs callable."""
fn = build_tool_fn(target, auth_resolver, name=name)
return Tool.from_function(fn, name=fn.__name__, tags=tags, annotations=annotations)
|