o i"@s,ddlmZddlZddlZddlZddlZddlZddlmZm Z m Z ddl m Z m Z mZddlmZmZmZddlmZmZddlZddlZddlmZddlmZdd lmZerbdd lmZe d3d dZ d4ddZ!e ddddd5d%d&Z"e  ' ( d6d7d-d.Z#e d/d0Z$Gd1d2d2eZ%dS)8) annotationsN)AsyncGeneratorCallable Generator)asynccontextmanagercontextmanagersuppress) TYPE_CHECKINGAnyLiteral)parse_qsurlparse)settings)OAuth)find_available_port)FastMCPkwargsr c ksrtt}z#|D] \}}t||q dVW|D] }t|||qdS|D] }t|||q,w)a Temporarily override FastMCP setting values. Args: **kwargs: The settings to override, including nested settings. Example: Temporarily override a setting: ```python import fastmcp from fastmcp.utilities.tests import temporary_settings with temporary_settings(log_level='DEBUG'): assert fastmcp.settings.log_level == 'DEBUG' assert fastmcp.settings.log_level == 'INFO' ``` N)copydeepcopyritems set_setting get_setting)r old_settingsattrvaluer]/var/www/html/karishye-ai-python/venv/lib/python3.10/site-packages/fastmcp/utilities/tests.pytemporary_settingss  r mcp_serverr transportLiteral['sse']portintreturnNonecCsL|dkr |jdd}ntd|tjtj|d|dddd}|dS) Nsse)rzInvalid transport: 127.0.0.1errorzwebsockets-sansio)apphostr! log_levelws)config)http_app ValueErroruvicornServerConfigrun)rrr!r(uvicorn_serverrrr _run_server9s r4Tr&)provide_host_and_portr)r! server_fnCallable[..., None]argsr5boolr)str int | NoneGenerator[str, None, None]c osb|durt}|r|||dO}tj|||dd}|d}d}||kr||r|z#ttjtj} | ||f WdWn@1sGwYWn&t ys|dkr^t dn|d krht d nt d |d 7}Ynw||kr||s*t d |dd|d|V| |jdd|r||jdd|rt ddSdS)a Context manager that runs a FastMCP server in a separate process and returns the server URL. When the context manager is exited, the server process is killed. Args: server_fn: The function that runs a FastMCP server. FastMCP servers are not pickleable, so we need a function that creates and runs one. *args: Arguments to pass to the server function. provide_host_and_port: Whether to provide the host and port to the server function as kwargs. host: Host to bind the server to (default: "127.0.0.1"). port: Port to bind the server to (default: find available port). **kwargs: Keyword arguments to pass to the server function. Returns: The server URL. N)r)r!T)targetr8rdaemonrg?皙?g?zServer failed to start after z attemptshttp://:)timeoutz2Server process failed to terminate even after kill)rmultiprocessingProcessstartis_alivesocketAF_INET SOCK_STREAMconnectConnectionRefusedErrortimesleep RuntimeError terminatejoinkill) r6r5r)r!r8rproc max_attemptsattemptsrrrrun_server_in_processKsJ$       r[http/mcpserver)Literal['http', 'streamable-http', 'sse']pathAsyncGenerator[str, None]cCsddl}|dur t}|dIdH||j||||dd}|dIdHz-d|d||VW|t|j|IdHWddS1sPwYdS|t|j |IdHWdw1spwYw) a_ Start a FastMCP server as an asyncio task for in-process async testing. This is the recommended way to test FastMCP servers. It runs the server as an async task in the same process, eliminating subprocess coordination, sleeps, and cleanup issues. Args: server: FastMCP server instance port: Port to bind to (default: find available port) transport: Transport type ("http", "streamable-http", or "sse") path: URL path for the server (default: "/mcp") host: Host to bind to (default: "127.0.0.1") Yields: Server URL string Example: ```python import pytest from fastmcp import FastMCP, Client from fastmcp.client.transports import StreamableHttpTransport from fastmcp.utilities.tests import run_server_async @pytest.fixture async def server(): mcp = FastMCP("test") @mcp.tool() def greet(name: str) -> str: return f"Hello, {name}!" async with run_server_async(mcp) as url: yield url async def test_greet(server: str): async with Client(StreamableHttpTransport(server)) as client: result = await client.call_tool("greet", {"name": "World"}) assert result.content[0].text == "Hello, World!" ``` rNg{Gz?F)r)r!rr` show_bannerrBrDrE)asynciorrR create_taskrun_http_asynccancelrCancelledError)r^r!rr`r)rc server_taskrrrrun_server_asyncs21   "  ric csH|td}||jz |VW||jdS||jw)zWContext manager to capture logs from FastMCP loggers even when propagation is disabled.fastmcpN)clearlogging getLogger addHandlerhandler removeHandler)caplogloggerrrrcaplog_for_fastmcps  rscs6eZdZdZdfdd Zdd d Zdd d ZZS) HeadlessOAuthz OAuth provider that bypasses browser interaction for testing. This simulates the complete OAuth flow programmatically by making HTTP requests instead of opening a browser and running a callback server. Useful for automated testing. mcp_urlr:c sd|_tj|fi|dS)z7Initialize HeadlessOAuth with stored response tracking.N)_stored_responsesuper__init__)selfrur __class__rrrxszHeadlessOAuth.__init__authorization_urlr#r$c s^t4IdH}|j|ddIdH}||_WdIdHdS1IdHs(wYdS)zOMake HTTP request to authorization URL and store response for callback handler.NF)follow_redirects)httpx AsyncClientgetrv)ryr|clientresponserrrredirect_handlers .zHeadlessOAuth.redirect_handlertuple[str, str | None]c s|jstd|j}|jdkrN|jd}t|}t|j}d|vr;|dd}|ddgd}td|d ||d d}|d d gd}||fStd |j)z4Parse stored response and return (auth_code, state).zHNo authorization response stored. redirect_handler must be called first.i.locationr'rerror_descriptionz Unknown errorzOAuth authorization failed: z - codestateNzAuthorization failed: )rvrS status_codeheadersr r queryr) ryr redirect_urlparsed query_paramsr' error_desc auth_coderrrrcallback_handlers(     zHeadlessOAuth.callback_handler)rur:)r|r:r#r$)r#r)__name__ __module__ __qualname____doc__rxrr __classcell__rrrzrrts  rt)rr )rrrr r!r"r#r$)r6r7r8r r5r9r)r:r!r;rr r#r<)Nr\r]r&) r^rr!r;rr_r`r:r)r:r#ra)& __future__rrrlrHrLrQcollections.abcrrr contextlibrrrtypingr r r urllib.parser r r~r/rjrfastmcp.client.auth.oauthrfastmcp.utilities.httprfastmcp.server.serverrrr4r[rirsrtrrrrsD       CO