Is anthropic safe?

return (path.name, path.read_bytes())

return pathlib.Path(file).read_bytes()

return (path.name, await path.read_bytes())

return await anyio.Path(file).read_bytes()

binary = data.read_bytes()

binary = await anyio.Path(data).read_bytes()

contents = Path(path).read_bytes()

files.append((path.relative_to(relative_to).as_posix(), path.read_bytes()))

files.append((path.relative_to(relative_to).as_posix(), await path.read_bytes()))

name = (_config_dir() / "active_config").read_text(encoding="utf-8").strip()

raw = self._config_path.read_text(encoding="utf-8")

creds: Dict[str, Any] = _wrap_secret_fields(json.loads(path.read_text(encoding="utf-8")))

content = self._path.read_text(encoding="utf-8").strip()

return full_path.read_text(encoding="utf-8")

return await full_path.read_text(encoding="utf-8")

data = base64.standard_b64encode(target.read_bytes()).decode("ascii")

text = target.read_text(encoding="utf-8")

text = target.read_text(encoding="utf-8")

with open(file, mode="wb") as f:

with open(file, mode="wb") as f:

with open(file, mode="wb") as f:

with open(file, mode="wb") as f:

os.unlink(tmp)

shutil.rmtree(full_path)

shutil.rmtree(self.memory_root)

shutil.rmtree(dest, ignore_errors=True)

os.unlink(dest)

os.unlink(tmp)

with zf.open(info) as src, open(target, "wb") as out:

shutil.copyfileobj(src, out)

with extracted as src, open(target, "wb") as out:

shutil.copyfileobj(src, out)

target.write_text(content, encoding="utf-8")

target.write_text(updated, encoding="utf-8")

import httpx

from httpx import URL, Proxy, HTTPTransport, AsyncHTTPTransport

from httpx._config import (
        DEFAULT_TIMEOUT_CONFIG,  # pyright: ignore[reportPrivateImportUsage]
    )

from httpx._config import DEFAULT_TIMEOUT_CONFIG as HTTPX_DEFAULT_TIMEOUT

return httpx.QueryParams(cast(Any, self._options.params)).merge(url.params)

self._base_url = self._enforce_trailing_slash(URL(base_url))

headers = httpx.Headers(headers_dict)

merge_url = URL(url)

self._base_url = self._enforce_trailing_slash(url if isinstance(url, URL) else URL(url))

proxy_map = {key: None if url is None else Proxy(url=url) for key, url in get_environment_proxies().items()}

key: None if proxy is None else HTTPTransport(proxy=proxy, **transport_kwargs)

default_transport = HTTPTransport(**transport_kwargs)

proxy_map = {key: None if url is None else Proxy(url=url) for key, url in get_environment_proxies().items()}

key: None if proxy is None else AsyncHTTPTransport(proxy=proxy, **transport_kwargs)

default_transport = AsyncHTTPTransport(**transport_kwargs)

import httpx

return requested is None or str(httpx.URL(requested)).rstrip("/") == str(current).rstrip("/")

import httpx

DEFAULT_TIMEOUT = httpx.Timeout(timeout=10 * 60, connect=5.0)

DEFAULT_CONNECTION_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100)

import httpx

import httpx

import httpx

from urllib.parse import parse_qs, urlencode

return parse_qs(query)

# by default, so new fields (id_token, client_secret, ...) are wrapped without

or the workload-identity env trio). Per the credential-precedence spec,

f"{ENV_IDENTITY_TOKEN} is not set; the workload-identity chain "

raise RuntimeError("Cannot call a synchronous function asynchronously. Use `@tool` instead.")