(model_name: str)
| 4594 | return main_cost, total_cost |
| 4595 | |
| 4596 | def _prepare_attempt(model_name: str) -> dict[str, Any]: |
| 4597 | attempt_provider_entry = _providers.get_for_model(model_name) |
| 4598 | attempt_upstream_body = json.loads(json.dumps(body)) |
| 4599 | _requested_max = attempt_upstream_body.get("max_tokens") |
| 4600 | if isinstance(_requested_max, int) and _requested_max > UPSTREAM_MAX_OUTPUT_TOKENS: |
| 4601 | attempt_upstream_body["max_tokens"] = UPSTREAM_MAX_OUTPUT_TOKENS |
| 4602 | attempt_headers: dict[str, str] = {} |
| 4603 | for key in ( |
| 4604 | "authorization", |
| 4605 | "content-type", |
| 4606 | "accept", |
| 4607 | "user-agent", |
| 4608 | _RECURSION_GUARD_HEADER, |
| 4609 | _ORIGINAL_MODEL_HEADER, |
| 4610 | ): |
| 4611 | val = request.headers.get(key) |
| 4612 | if val: |
| 4613 | attempt_headers[key] = val |
| 4614 | if api_format == "anthropic" and "authorization" not in attempt_headers: |
| 4615 | x_api_key = request.headers.get("x-api-key") |
| 4616 | if x_api_key: |
| 4617 | attempt_headers["authorization"] = f"Bearer {x_api_key}" |
| 4618 | if "content-type" not in attempt_headers: |
| 4619 | attempt_headers["content-type"] = "application/json" |
| 4620 | attempt_headers["user-agent"] = f"uncommon-route/{VERSION}" |
| 4621 | if is_virtual and not attempt_provider_entry: |
| 4622 | attempt_headers[_RECURSION_GUARD_HEADER] = "1" |
| 4623 | if requested_model: |
| 4624 | attempt_headers[_ORIGINAL_MODEL_HEADER] = requested_model |
| 4625 | |
| 4626 | resolved_model = model_name |
| 4627 | if not attempt_provider_entry: |
| 4628 | resolved_model = _mapper.resolve(model_name) |
| 4629 | attempt_upstream_body["model"] = resolved_model |
| 4630 | |
| 4631 | attempt_has_tools = bool( |
| 4632 | attempt_upstream_body.get("tools") |
| 4633 | or attempt_upstream_body.get("customTools") |
| 4634 | ) |
| 4635 | attempt_has_tool_results = any( |
| 4636 | isinstance(message, dict) and str(message.get("role", "")).strip().lower() == "tool" |
| 4637 | for message in attempt_upstream_body.get("messages", []) |
| 4638 | if isinstance(message, dict) |
| 4639 | ) |
| 4640 | attempt_transport_decision = _choose_transport( |
| 4641 | api_format=api_format, |
| 4642 | endpoint_name=endpoint_name, |
| 4643 | selected_model=model_name, |
| 4644 | provider_entry=attempt_provider_entry, |
| 4645 | upstream_provider=_mapper.provider, |
| 4646 | upstream_base=upstream, |
| 4647 | step_type=step_type, |
| 4648 | has_tools=attempt_has_tools, |
| 4649 | has_tool_results=attempt_has_tool_results, |
| 4650 | anthropic_beta_present=bool(request.headers.get("anthropic-beta")), |
| 4651 | ) |
| 4652 | attempt_native_anthropic_transport = attempt_transport_decision.native_anthropic_transport |
| 4653 | if attempt_native_anthropic_transport: |
no test coverage detected