kKjjrhUdZddlZddlZddlZddlZddlmZddlmZm Z ddl m Z ddl Z ddl ZddlZddlmZddlmZddlmZmZdd lmZdd lmZdd lmZed ZeZhd Zdee e fde!fdZ"dee e fde#e!e dzffdZ$dee e fde#e e!ffdZ%dee e fde fdZ&de ee e fzezde#e e'e ffdZ(de de fdZ)dejTjVde fdZ,de de de de'e de-ejTjVde f dZ.d d!d"d#d$d%d&d'Z/ee e fe0d(<d)e de fd*Z1d+ee e fde fd,Z2dejTjVd-e de fd.Z3de d-e de-ejTjVde fd/Z4 dFd0ddd0d1dee e jjd23fd4ee e jjd53fd6ee!e jjd7d8gd9:fd;ee6dze jjdee dze jjd?d@3fdAee!e jjdBdC3fddfdDZ7de de fdEZ8y)GzDGenerate a standalone CLI script and agent skill from an MCP server.N)Path) AnnotatedAny)urlparse)McpError)Console) _build_clientresolve_server_spec)ClientTransport)StdioTransport) get_loggerz cli.generate>nullnumberstringbooleanintegerschemareturncx|jd}t|trtd|DS|tvS)z9Check if a schema represents a simple (non-complex) type.typec3,K|] }|tvywN) _SIMPLE_TYPES).0ts e/Users/ahmed/devFolder/Ultron/claude-voice/.venv/lib/python3.12/site-packages/fastmcp/cli/generate.py z"_is_simple_type..%s;{!1 %{s)get isinstancelistallr)r schema_types r_is_simple_typer# s6**V$K+t$;{;;; - ''c|jddk7ry|jdi}t|sy|jdd}t|trydddd d }|j|}|yd |fS) z`Check if schema is an array of simple types. Returns (is_simple_array, item_type_str). rarray)FNitemsrstrintfloatbool)rrrrT)rr#rr )rr' item_typetype_mappy_types r_is_simple_arrayr/)s zz&W$ JJw #E 5 ! &(+I)T" H ll9%G =r$cDt|\}}|rd|ddfSt|rx|jdd}t|tr:dddd d d }|Dcgc]}|j|d}}d j |dfSdddd d d }|j|ddfSy cc}w)zkConvert a JSON Schema to a Python type annotation. Returns (type_annotation, needs_json_parsing). list[]Frrr(r)r*r+None)rrrrr | )r(T)r/r#rrr join)r is_simple_arrr,r"r-rpartss r_schema_to_python_typer8Es 07M9yk#U**vjj2 k4 (  !! H6AA[X\\!U+[EA::e$e+ +  ||K/66 BsBctj|dj}|jd}dj |}d|S)z.Format a JSON schema for display in help text.)indent z z JSON Schema: ) pydantic_coreto_jsondecodesplitr5)r schema_strlinesindenteds r_format_schema_for_helprDksL&&va8??AJ   T "E-2259H 8* %%r$resolvedct|trt|tfSt|trd|j d|j g}|jr|jd|j|jr|jd|jddj|d}dh}||fSt|trt|tfSt|tfS) zSerialize a resolved transport to a Python expression string. Returns ``(expression, extra_imports)`` where *extra_imports* is a set of import lines needed by the expression. zcommand=zargs=zenv=zcwd=zStdioTransport(, )z4from fastmcp.client.transports import StdioTransport) rr(reprsetr commandargsenvappendcwdr5dict)rEr7exprimportss rserialize_transportrSzs(C H~su$$(N+H,,/0E(--9J2KL << LL4 /0 1 << LL4 /0 1 5!1 2!4IJW}(D!H~su$$ >35  r$namectjdd|}|r|djrd|}|xsd}tj|r|d}|S)z1Sanitize a string into a valid Python identifier.z [^a-zA-Z0-9_]_r_unnamed)resubisdigitkeyword iskeyword)rTsafes r_to_python_identifierr^sX 66"C .D Q!4&z  :Dqz Kr$toolc |j}|jdi}t|jdg}g}g}g}i}|jD]\}t |\} } |jdd} |v} t } | |vrt dd|| d| d|| <| r)t|}| r| d |n|} |j| f| jd d jd d jdd }| r"d| d|d}|jd| d|dn|jd}|u| rNtj|tj}d| d|d}|jd| d|d|dnyd| d|d}|jd| d|d|dnT| jdr"d| d|d}|jd| d|dn!d| d|d}|jd| d|d|jd| t |j}|j xsdjd d jdd}g}|jd|jd|jd|jd|d |r"|jd!|j#||jd"|jd#|d$|rI|jd%|D]"\}} |jd| d&| d'| d(| $|jdg}|D]Ot } t%fd)|Dr|jd| d*:|jd| Qd+j'|}|jd,|jd-|d.|jddj'|S)/zEGenerate the source for a single ``@call_tool_app.command`` function. propertiesrequired descriptionzParameter name collision: 'z' and 'z' both sanitize to ''z\n\\\"\"r<z Annotated[z, cyclopts.Parameter(help="z")]z z: ,default)fallbackz = r1z = [],z" | None, cyclopts.Parameter(help="z = None,z\'z@call_tool_app.command(name=rHz async def (z *,z ) -> None:z '''z'''z # Parse JSON parametersz_parsed = json.loads(z) if isinstance(z , str) else c3.K|] \}}|k(ywr)rpnrV prop_names rrz(_tool_function_source.. s8K52qrYKs_parsedrGz await _call_tool(z, {z})) inputSchemarrJr'r8r^ ValueErrorrDrNreplacer=r>r(r? startswithrTrcextendanyr5)r_rrarb param_lines call_args json_params seen_names prop_schemar. needs_json help_text is_required safe_name schema_help help_escaped annotationrk default_strfn_namercrB _prop_namecall_arg_parts dict_itemsrqs @r_tool_function_sourcers   F!'L"!=J6::j"-.H KI)+K!#J","2"2"4 ;4[AOOM26 8+ ))4   "-i[ 9@U?VW%%.Kq2 !* 9 1+>K:C9+S 6I    95 6   dF + 3 3C ? G Ge T  WI%@cR    i[:,a@ A!ooi0G""/"7"7#"N"U"U"WK#-gY6QR^Q__b!cJ&&ykJ? LL:gYa() X [! LL LL7;-s+, 23%0 !J LLyk!6ykAQR[Q\\hirhst &1  RN ))4 8K8 8  ! !YMI;g"F G  ! !YMI;"? @  >*J LL( T*SIJ LL 99U r$ server_name server_spectransport_code extra_importstoolsc|jddjjddjdd}g}|jd|jd|d |jd |jd ||jd |jd |jd |jd|jd|jd |jd|jd|jd|jd |jd|jt ||jd |jd|jd||jd |jddjdd}|jd|d|d|jd|jd|jd |jd|jd |jd |jt j d|jd |jd |jt j d|jd |jd |rK|jd|jd|jd|jd |D|jd |jd!|jd"|jd d#j|S)$z)Generate the full CLI script source code. -rfrgrhriz#!/usr/bin/env python3z """CLI for z MCP server.rdz#Generated by: fastmcp generate-cli z"""z import jsonz import syszfrom typing import Annotatedzimport cycloptszimport mcp.typesz from rich.console import Consolezfrom fastmcp import Clientz?# Modify this to change how the CLI connects to the MCP server.zCLIENT_SPEC = zapp = cyclopts.App(name="z", help="CLI for z MCP server")zPcall_tool_app = cyclopts.App(name="call-tool", help="Call a tool on the server")zapp.command(call_tool_app)zconsole = Console()ao # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _print_tool_result(result): if result.is_error: for block in result.content: if isinstance(block, mcp.types.TextContent): console.print(f"[bold red]Error:[/bold red] {block.text}") else: console.print(f"[bold red]Error:[/bold red] {block}") sys.exit(1) if result.structured_content is not None: console.print_json(json.dumps(result.structured_content)) return for block in result.content: if isinstance(block, mcp.types.TextContent): console.print(block.text) elif isinstance(block, mcp.types.ImageContent): size = len(block.data) * 3 // 4 console.print(f"[dim][Image: {block.mimeType}, ~{size} bytes][/dim]") elif isinstance(block, mcp.types.AudioContent): size = len(block.data) * 3 // 4 console.print(f"[dim][Audio: {block.mimeType}, ~{size} bytes][/dim]") async def _call_tool(tool_name: str, arguments: dict) -> None: # Filter out None values and empty lists (defaults for optional array params) filtered = { k: v for k, v in arguments.items() if v is not None and (not isinstance(v, list) or len(v) > 0) } async with Client(CLIENT_SPEC) as client: result = await client.call_tool(tool_name, filtered, raise_on_error=False) _print_tool_result(result) if result.is_error: sys.exit(1)u # --------------------------------------------------------------------------- # List / read commands # --------------------------------------------------------------------------- @app.command async def list_tools() -> None: """List available tools.""" async with Client(CLIENT_SPEC) as client: tools = await client.list_tools() if not tools: console.print("[dim]No tools found.[/dim]") return for tool in tools: sig_parts = [] props = tool.inputSchema.get("properties", {}) required = set(tool.inputSchema.get("required", [])) for pname, pschema in props.items(): ptype = pschema.get("type", "string") if pname in required: sig_parts.append(f"{pname}: {ptype}") else: sig_parts.append(f"{pname}: {ptype} = ...") sig = f"{tool.name}({', '.join(sig_parts)})" console.print(f" [cyan]{sig}[/cyan]") if tool.description: console.print(f" {tool.description}") console.print() @app.command async def list_resources() -> None: """List available resources.""" async with Client(CLIENT_SPEC) as client: resources = await client.list_resources() if not resources: console.print("[dim]No resources found.[/dim]") return for r in resources: console.print(f" [cyan]{r.uri}[/cyan]") desc_parts = [r.name or "", r.description or ""] desc = " — ".join(p for p in desc_parts if p) if desc: console.print(f" {desc}") console.print() @app.command async def read_resource(uri: Annotated[str, cyclopts.Parameter(help="Resource URI")]) -> None: """Read a resource by URI.""" async with Client(CLIENT_SPEC) as client: contents = await client.read_resource(uri) for block in contents: if isinstance(block, mcp.types.TextResourceContents): console.print(block.text) elif isinstance(block, mcp.types.BlobResourceContents): size = len(block.blob) * 3 // 4 console.print(f"[dim][Blob: {block.mimeType}, ~{size} bytes][/dim]") @app.command async def list_prompts() -> None: """List available prompts.""" async with Client(CLIENT_SPEC) as client: prompts = await client.list_prompts() if not prompts: console.print("[dim]No prompts found.[/dim]") return for p in prompts: args_str = "" if p.arguments: parts = [a.name for a in p.arguments] args_str = f"({', '.join(parts)})" console.print(f" [cyan]{p.name}{args_str}[/cyan]") if p.description: console.print(f" {p.description}") console.print() @app.command async def get_prompt( name: Annotated[str, cyclopts.Parameter(help="Prompt name")], *arguments: str, ) -> None: """Get a prompt by name. Pass arguments as key=value pairs.""" parsed: dict[str, str] = {} for arg in arguments: if "=" not in arg: console.print(f"[bold red]Error:[/bold red] Invalid argument {arg!r} — expected key=value") sys.exit(1) key, value = arg.split("=", 1) parsed[key] = value async with Client(CLIENT_SPEC) as client: result = await client.get_prompt(name, parsed or None) for msg in result.messages: console.print(f"[bold]{msg.role}:[/bold]") if isinstance(msg.content, mcp.types.TextContent): console.print(f" {msg.content.text}") elif isinstance(msg.content, mcp.types.ImageContent): size = len(msg.content.data) * 3 // 4 console.print(f" [dim][Image: {msg.content.mimeType}, ~{size} bytes][/dim]") else: console.print(f" {msg.content}") console.print()zM# ---------------------------------------------------------------------------z.# Tool commands (generated from server schema)c32K|]}t|ywr)r)rr_s rrz&generate_cli_script..sCUT*40Uszif __name__ == "__main__":z app()r<)rulowerrNrwsortedtextwrapdedentr5)rrrrrapp_namerBserver_name_escapeds rgenerate_cli_scriptrs C%++-55dFCKKCQVW E LL)* LL;{m<89 LL LL6{mDE LL LL LL LL LL/0 LL LL"# LL#$ LL34 LL LL-. LL &' LL LLRS LL>.!123 LL&--dF;CCCO LL #H:->?R>SS`a LLZ LL-. LL LL&' LL LL LL)#) $+X LL LL LLi'i (kX LL LL  [   EF [   CUCC LL LL-. LL LL 99U r$rrrrrr&object)rrrrrr&r_JSON_SCHEMA_TYPE_LABELSrqct|}tjdd|}tjdd|}|jj ddj d}|rd|SdS)uConvert a JSON Schema property name to its CLI flag form. Replicates cyclopts' default_name_transform: camelCase → snake_case, lowercase, underscores → hyphens, strip leading/trailing hyphens. z(.)([A-Z][a-z]+)z\1_\2z([a-z0-9])([A-Z])rVrz--z--arg)r^rXrYrrustrip)rqr]s r_param_to_cli_flagrsj ! +D 66%x 6D 66&$ 7D ::<  S ) / / 4DRv;+G+r$r}cv|jdd}t|tr4|Dcgc]}tj||}}dj |Stj||}|dk(rB|jdi}|jdd}t|t r|tvrd|dS|Scc}w) z9Return a human-readable type label for a property schema.rrr4r&r'rdzarray[r2)rrr rr5r()r}r"rlabelslabelr'r,s r_schema_type_labelr$s//&(3K+t$>IJk*..q!4kJzz&!! $ ( (k BEg,IIfb) i %)7O*OI;a( ( LKsB6 cli_filenamec z|j}|jdi}t|jdg}g}|jD]f\}}t |}|jd} | dk(xst | t xrd| v} | r|j|S|j|dhdj|} d|d|j} | r| d| z } g} |jD]\}}d t |d }t|jd d }||vrd nd }|jdd}t|\}}|r |r|dnd}|jddjd d }| jd|d|d|d|d d}| rd}d|ddj| zdz}d|jg}|jr|jd|jg|jdd| dg|r"|jd|jdgdj|S)z.Generate a SKILL.md section for a single tool.rarbrrz ruv run --with fastmcp python z call-tool `|z\|yesnorcrdz (JSON string)z JSON stringr<z| r4z |zQ| Flag | Type | Required | Description | |------|------|----------|-------------|z### ```bash```)rsrrJr'rrr rNr5rTrrur8rcrwr)r_rrrarbflag_parts_listpp_schemaflagr"is_bool flag_parts invocationrowsrqr} type_labelrrcrVr~ param_tableheaderrBs r_tool_skill_sectionr7si   F!'L"!=J6::j"-.H"$O!'') 8!!$ll6* * {D ) Fi;.F    " "4 (  " "dV8#4 5*/*J0k$))UJ*&& D","2"2"4 ;%i013' 4< tyyk*+E  b$**+, LL"iU34 b+++D123 99U r$c |jddjjddjdd}|jddjdd}d|d}dd |d d |dddd |d dg}|r/djfd|D}|jdd|dg|jdddddddddddddddg dj|S)z4Generate a SKILL.md file for a generated CLI script.rrrfrdrhz CLI for the z9 MCP server. Call tools, list resources, and get prompts.z---zname: "z-cli"zdescription: "z# z CLIz c36K|]}t|ywr)r)rr_rs rrz)generate_skill_content..s" @E l 3sz## Tool Commandsz## Utility Commandsrrz list-toolsz list-resourcesz read-resource z list-promptsz" get-prompt [key=value ...]rr<)rurr5rw)rrr skill_namerrcrB tool_bodiess ` rgenerate_skill_contentrksM C%++-55dB?GGRP##D"-55c2>I  +deK  *U#  Q'  [M  E kk" @E"    ("k2>? LL !  +L> E+L> I+L>9M N+L> G+L>9[ \     99U r$F)forcetimeoutauthno_skillzEServer URL, Python file, MCPConfig JSON, discovered name, or .js file)helpoutputz"Output file path (default: cli.py)rz-fz--forcez"Overwrite output file if it exists)rTrrz --timeoutzConnection timeout in secondsrz--authzAAuth method: 'oauth', a bearer token string, or 'none' to disablerz --no-skillz8Skip generating a SKILL.md agent skill alongside the CLIcKt|}|jdz }g}|jr|s|j||s#|jr|s|j||rFdj d|D} t j d| dtjdt|} t| \} } t|} t| ||} |4d{|jd{}t j d t|d |d dddd{t'| || |  }|j)||j+|j-j.dzt j d|dt|d|sBt1| |j2|}|j)|t j d|dt j d|dy7"7 7#1d{7swYxYw#tt t"t$f$r8}t j d |tjdYd}~9d}~wwxYww)aGenerate a standalone CLI script from an MCP server. Connects to the server, reads its tools/resources/prompts, and writes a Python script that can invoke them directly. Also generates a SKILL.md agent skill file unless --no-skill is passed. Examples: fastmcp generate-cli weather fastmcp generate-cli weather my_cli.py fastmcp generate-cli http://localhost:8000/mcp fastmcp generate-cli server.py output.py -f fastmcp generate-cli weather --no-skill zSKILL.mdrGc3(K|] }d|d yw)z[cyan][/cyan]Nro)rrs rrz'generate_cli_command..s@x!F1#W-xsz[bold red]Error:[/bold red] z4 already exist(s). Use [cyan]-f[/cyan] to overwrite.)rrNz[dim]Discovered z tool(s) from z[/dim]z/[bold red]Error:[/bold red] Could not connect: )rrrrrIu[green]✓[/green] Wrote [cyan]z [/cyan] with z tool command(s))rrrrz[dim]Run: python z --help[/dim])rparentexistsrNr5consoleprintsysexitr rS_derive_server_namer list_toolslen RuntimeError TimeoutErrorrOSErrorr write_textchmodstatst_moderrT)rrrrrr output_path skill_pathexistingnamesrErrrclientrexcscript skill_contents rgenerate_cli_commandrs=jv,K##j0JHE $  ))+E # @x@@ *5'20 1   #;/H$7$A!NM&k2K8W4 @F 6 ++--E MM"3u:,n[MP 6!%# F6"k&&(00589 MM )+7E |+ - .#$)) m, 7 |7KL MM%k]-@AI-666 ,' : GuMN  sCI;"H'(H )H',HH )H* H'5H6H':CI; H' HH'H$H H$ H''I8-I3-I;3I88I;c|jdrt|}|jxsdS|jdrt |j Sd|vr.|j ddd}|xs|j dddS|S)z0Derive a human-friendly name from a server spec.)zhttp://zhttps://server)z.pyz.jsz.json:rr)rvrhostnameendswithrstemr@)rparsedrTs rrrs56+&*(*34K %%% k  a(+3{((a033 r$)zcli.py)9__doc__r[rXrrpathlibrtypingrr urllib.parsercyclopts mcp.typesmcpr=r rich.consolerfastmcp.cli.clientr r fastmcp.client.transports.baser fastmcp.client.transports.stdior fastmcp.utilities.loggingr loggerrrrPr(r+r#tupler/r8rDrJrSr^typesToolrr rr__annotations__rrrr Parameterr*rrror$rrsJ !! A::0 N # ) C (DcN(t(T#s(^dC$J6F0G8#4S>#eCI6F#L&DcN&s&!DcN"_4! 3C=!Bm m3mjiiiis8 i    i  ib  ,$sCx. ,# ,# ,DcNs&1ciinn1C1C1h+++   + +~ uB(    MuB X  uB  5   uB   "5   uB*  ;-LM O+uB2  d  T   3uB@  K  AuBN OuBpSSr$