> For the complete documentation index, see [llms.txt](https://fastapi-docs.duonghuuphuc.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://fastapi-docs.duonghuuphuc.com/unit-20-integrating-third-party-services.md).

# Unit 20: Integrating Third-Party Services

Integrate a FastAPI service with a mock third-party provider using REST APIs, API keys, error handling, idempotency, and webhooks.

### 1. Scenario

Modern applications frequently depend on external services such as:

* Email providers
* Payment gateways
* Delivery services
* Maps and location APIs
* AI services
* Cloud storage services

Instead of implementing all functionality ourselves, we integrate with a third-party provider through its API.

In this unit, we will build:

* **Notification Service** — our application service.
* **Mock Email Provider** — simulates an external email API.

The architecture is:

```
Client
   │
   │ POST /notifications/email
   ▼
Notification Service :8000
   │
   │ HTTPS / REST
   │ API Key
   ▼
Mock Email Provider :9000
```

The mock provider behaves like a real external service but runs locally.

This allows us to practice:

* Sending requests to an external API
* Using API keys
* Mapping internal data to provider-specific payloads
* Processing responses
* Handling errors
* Handling rate limits
* Using idempotency
* Receiving webhooks

***

## 2. Project Structure

Create the following structure:

```
unit20-third-party/
│
├── notification_service/
│   └── main.py
│
└── mock_email_provider/
    └── main.py
```

The two applications run independently.

```
Notification Service   → Port 8000
Mock Email Provider    → Port 9000
```

***

## 3. Step 1 — Build the Mock Email Provider

Open:

```
mock_email_provider/main.py
```

Create the application:

```python
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from uuid import uuid4

app = FastAPI(title="Mock Email Provider")
```

Define the API key expected by the provider:

```python
VALID_API_KEY = "demo-secret-key"
```

Later, our Notification Service must include this key in every request.

***

## 4. Step 2 — Define the Provider Request Model

Add:

```python
class ProviderEmailRequest(BaseModel):
    to: str
    subject: str
    content: str
```

The mock provider expects requests in this format:

```json
{
    "to": "student@example.com",
    "subject": "Order Confirmation",
    "content": "Your order has been received."
}
```

Notice the provider-specific field names:

```
to
subject
content
```

Our own Notification Service does not necessarily need to use the same model.

***

## 5. Step 3 — Create the Provider Endpoint

Add:

```python
@app.post("/api/send")
def send_email(
    email: ProviderEmailRequest,
    x_api_key: str = Header()
):
    if x_api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    message_id = str(uuid4())

    return {
        "message_id": message_id,
        "status": "accepted",
        "provider": "MockMail"
    }
```

The provider now exposes:

```
POST /api/send
```

and requires:

```
X-API-Key
```

Example request:

```
POST /api/send
X-API-Key: demo-secret-key
```

Body:

```json
{
    "to": "student@example.com",
    "subject": "Hello",
    "content": "Welcome to our service."
}
```

Example response:

```json
{
    "message_id": "9de7427c-5e85-4d07-9765-4545f4eeb336",
    "status": "accepted",
    "provider": "MockMail"
}
```

***

## 6. Step 4 — Run the Mock Provider

Run:

```bash
uvicorn main:app --reload --port 9000
```

The provider is available at:

```
http://127.0.0.1:9000
```

Test:

```
POST /api/send
```

Try both:

```
X-API-Key: demo-secret-key
```

and an invalid key:

```
X-API-Key: wrong-key
```

The invalid request should return:

```
401 Unauthorized
```

***

## 7. Step 5 — Build the Notification Service

Open:

```
notification_service/main.py
```

Add:

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
import os

app = FastAPI(title="Notification Service")
```

Define the provider URL:

```python
EMAIL_PROVIDER_URL = "http://127.0.0.1:9000"
```

Read the API key from an environment variable:

```python
EMAIL_API_KEY = os.getenv(
    "EMAIL_API_KEY",
    "demo-secret-key"
)
```

For this tutorial, the default value allows the service to run immediately.

In production systems, secrets should not normally be hard-coded directly into application source code.

***

## 8. Step 6 — Define Our Internal Request Model

Create:

```python
class EmailNotification(BaseModel):
    recipient: str
    subject: str
    message: str
```

Our API accepts:

```json
{
    "recipient": "student@example.com",
    "subject": "Order Confirmation",
    "message": "Your order has been received."
}
```

Compare this with the provider model:

```
Our Model                 Provider Model

recipient        ───────► to
subject          ───────► subject
message          ───────► content
```

This separation is important.

Our internal API should not be unnecessarily coupled to the specific naming conventions of a third-party provider.

***

## 9. Step 7 — Map Internal Data to Provider Data

Create a helper function:

```python
def build_provider_payload(
    notification: EmailNotification
):
    return {
        "to": notification.recipient,
        "subject": notification.subject,
        "content": notification.message
    }
```

This function performs the mapping:

```
Internal Model
      ↓
Transformation
      ↓
Provider Payload
```

This design becomes useful if we later change providers.

For example:

```
Notification Service
       │
       ├── MockMail
       ├── SendGrid
       ├── Mailgun
       └── Another Provider
```

Our public API does not need to change.

***

## 10. Step 8 — Call the Third-Party API

Implement:

```python
@app.post("/notifications/email")
def send_notification(
    notification: EmailNotification
):
    payload = build_provider_payload(notification)

    headers = {
        "X-API-Key": EMAIL_API_KEY
    }

    response = httpx.post(
        f"{EMAIL_PROVIDER_URL}/api/send",
        json=payload,
        headers=headers
    )

    return response.json()
```

The communication flow is:

```
Client
   │
   │ POST /notifications/email
   ▼
Notification Service
   │
   │ Transform request
   │
   │ POST /api/send
   ▼
Mock Email Provider
```

***

## 11. Step 9 — Run Both Services

Keep the mock provider running on port `9000`.

Run Notification Service on port `8000`:

```bash
uvicorn main:app --reload --port 8000
```

We now have:

```
Notification Service
http://127.0.0.1:8000

Mock Email Provider
http://127.0.0.1:9000
```

Test:

```
POST /notifications/email
```

Request:

```json
{
    "recipient": "student@example.com",
    "subject": "Order Confirmation",
    "message": "Your order has been received."
}
```

The expected response is similar to:

```json
{
    "message_id": "6d7c3655-87cd-4c21-b27c-74558f36f351",
    "status": "accepted",
    "provider": "MockMail"
}
```

***

## 12. Step 10 — Understand the Integration Flow

The complete outbound flow is:

```
1. Client
      │
      │ POST /notifications/email
      ▼
2. Notification Service
      │
      │ Validate internal request
      │ Map internal fields
      │ Add API key
      ▼
3. Mock Email Provider
      │
      │ Validate API key
      │ Process request
      ▼
4. Provider Response
      │
      ▼
5. Notification Service
      │
      ▼
6. Client
```

This is a typical third-party API integration pattern.

***

## 13. API Keys

Many external services use API keys to identify applications.

Our provider expects:

```
X-API-Key: demo-secret-key
```

The Notification Service sends:

```python
headers = {
    "X-API-Key": EMAIL_API_KEY
}
```

The request becomes conceptually:

```
POST /api/send

Headers:
X-API-Key: demo-secret-key

Body:
{
    ...
}
```

If the API key is invalid, the provider returns:

```
401 Unauthorized
```

***

## 14. Environment Variables

Sensitive configuration should normally be separated from application code.

For example:

```bash
export EMAIL_API_KEY="demo-secret-key"
```

Then Python can read:

```python
EMAIL_API_KEY = os.getenv("EMAIL_API_KEY")
```

The same principle applies to:

```
PAYMENT_API_KEY
DELIVERY_API_KEY
AI_API_KEY
DATABASE_PASSWORD
```

Do not commit real API keys or passwords to a public source-code repository.

***

## 15. Step 11 — Handle Provider Errors

Our current implementation assumes that the provider always succeeds.

We should handle unsuccessful responses.

Replace:

```python
response = httpx.post(
    f"{EMAIL_PROVIDER_URL}/api/send",
    json=payload,
    headers=headers
)
```

with:

```python
try:
    response = httpx.post(
        f"{EMAIL_PROVIDER_URL}/api/send",
        json=payload,
        headers=headers,
        timeout=5.0
    )

except httpx.RequestError:
    raise HTTPException(
        status_code=503,
        detail="Email provider is unavailable"
    )
```

Now the service handles connection failures and timeouts.

***

## 16. Step 12 — Process HTTP Status Codes

Add:

```python
if response.status_code == 401:
    raise HTTPException(
        status_code=502,
        detail="Email provider authentication failed"
    )

if response.status_code == 429:
    raise HTTPException(
        status_code=503,
        detail="Email provider rate limit exceeded"
    )

if response.status_code >= 500:
    raise HTTPException(
        status_code=502,
        detail="Email provider error"
    )

if response.status_code != 200:
    raise HTTPException(
        status_code=400,
        detail="Email request rejected"
    )
```

Then:

```python
return response.json()
```

A third-party integration should not blindly assume every HTTP request succeeds.

***

## 17. Common Provider Responses

Typical third-party API responses include:

| Status                  | Meaning                        |
| ----------------------- | ------------------------------ |
| `200 OK`                | Request processed successfully |
| `400 Bad Request`       | Invalid request                |
| `401 Unauthorized`      | Invalid credentials            |
| `403 Forbidden`         | Operation not allowed          |
| `404 Not Found`         | Resource not found             |
| `429 Too Many Requests` | Rate limit exceeded            |
| `500+`                  | Provider-side error            |

The exact meaning depends on the provider's API documentation.

***

## 18. Rate Limiting

External providers often limit the number of requests clients can make.

For example:

```
100 requests / minute
```

If the limit is exceeded, the provider may return:

```
429 Too Many Requests
```

Modify the mock provider to simulate this condition.

Add:

```python
request_count = 0
```

Then modify the endpoint:

```python
@app.post("/api/send")
def send_email(
    email: ProviderEmailRequest,
    x_api_key: str = Header()
):
    global request_count

    if x_api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    request_count += 1

    if request_count > 5:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded"
        )

    message_id = str(uuid4())

    return {
        "message_id": message_id,
        "status": "accepted",
        "provider": "MockMail"
    }
```

After five requests, the mock provider starts returning:

```
429 Too Many Requests
```

This allows us to test how our application behaves when a provider applies a rate limit.

***

## 19. Idempotency

Consider the following situation:

```
Notification Service
       │
       │ Send email
       ▼
Provider
       │
       │ Email accepted
       ▼

Network response lost
```

Notification Service does not know whether the provider processed the request.

If it retries:

```
Send again
```

the provider may process the same operation twice.

To reduce this risk, many APIs support an **idempotency key**.

***

## 20. Step 13 — Add an Idempotency Key

Modify Notification Service:

```python
from uuid import uuid4
```

Generate a key:

```python
idempotency_key = str(uuid4())
```

Add it to the headers:

```python
headers = {
    "X-API-Key": EMAIL_API_KEY,
    "Idempotency-Key": idempotency_key
}
```

The request becomes:

```
POST /api/send

X-API-Key: demo-secret-key
Idempotency-Key: 3f755764-...
```

***

## 21. Step 14 — Handle Idempotency in the Provider

In the mock provider, create:

```python
processed_requests = {}
```

Modify the endpoint:

```python
@app.post("/api/send")
def send_email(
    email: ProviderEmailRequest,
    x_api_key: str = Header(),
    idempotency_key: str = Header()
):
    if x_api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    if idempotency_key in processed_requests:
        return processed_requests[idempotency_key]

    result = {
        "message_id": str(uuid4()),
        "status": "accepted",
        "provider": "MockMail"
    }

    processed_requests[idempotency_key] = result

    return result
```

Now:

```
Same Idempotency-Key
        ↓
Same operation result
        ↓
No duplicate processing
```

Idempotency is especially important for operations such as:

* Payments
* Order creation
* Booking
* Account transactions

***

## 22. Webhooks

So far, communication has been:

```
Our Service ───────► Third-Party Provider
```

However, third-party providers sometimes need to notify us later.

For example:

```
Email accepted
      ↓
Email delivered later
      ↓
Provider notifies our system
```

This is commonly implemented using a **webhook**.

The communication direction becomes:

```
Third-Party Provider
        │
        │ HTTP POST
        ▼
Our Service
```

***

## 23. Step 15 — Create a Webhook Endpoint

In Notification Service, define:

```python
class EmailStatusEvent(BaseModel):
    message_id: str
    status: str
```

Add:

```python
@app.post("/webhooks/email-status")
def receive_email_status(
    event: EmailStatusEvent
):
    print(
        f"Email {event.message_id}: "
        f"{event.status}"
    )

    return {
        "received": True
    }
```

The provider can now call:

```
POST /webhooks/email-status
```

with:

```json
{
    "message_id": "msg-123",
    "status": "delivered"
}
```

***

## 24. Outbound vs. Inbound Integration

We now have two communication directions.

### Outbound Integration

Our application calls the provider:

```
Notification Service
        │
        │ POST /api/send
        ▼
Email Provider
```

### Inbound Integration

The provider calls our application:

```
Email Provider
        │
        │ POST /webhooks/email-status
        ▼
Notification Service
```

Together:

```
             Outbound
Notification Service ─────────► Provider

             Inbound
Notification Service ◄───────── Provider
               Webhook
```

***

## 25. Callback vs. Webhook

The terms are sometimes used similarly, but conceptually:

**Callback**

A provider sends information back to a URL supplied by the client as part of an operation.

**Webhook**

A provider sends event notifications to a configured endpoint when particular events occur.

Examples:

```
Payment completed
Email delivered
Package shipped
Account updated
```

The exact behavior depends on the provider.

***

## 26. Sandbox vs. Production

Real providers often provide separate environments.

### Sandbox

Used for development and testing.

```
https://sandbox.provider.example
```

Typical characteristics:

* Test credentials
* Simulated transactions
* No real financial or operational effects
* Safe for development

### Production

Used by real applications.

```
https://api.provider.example
```

Typical characteristics:

* Production credentials
* Real transactions
* Real customers
* Real business effects

Conceptually:

```
Development
     │
     ▼
Sandbox API


Production Application
     │
     ▼
Production API
```

Never use production credentials for classroom experiments.

***

## 27. OAuth and Bearer Tokens

Not every provider uses API keys.

Some providers use OAuth access tokens.

Instead of:

```python
headers = {
    "X-API-Key": API_KEY
}
```

the request may use:

```python
headers = {
    "Authorization": f"Bearer {ACCESS_TOKEN}"
}
```

Conceptually:

```
API Key
    → identifies an application

OAuth Access Token
    → provides authorized access
```

A complete OAuth authorization flow is outside the scope of this unit.

***

## 28. Complete Mock Email Provider

`mock_email_provider/main.py`

```python
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from uuid import uuid4


app = FastAPI(title="Mock Email Provider")


VALID_API_KEY = "demo-secret-key"

processed_requests = {}


class ProviderEmailRequest(BaseModel):
    to: str
    subject: str
    content: str


@app.post("/api/send")
def send_email(
    email: ProviderEmailRequest,
    x_api_key: str = Header(),
    idempotency_key: str = Header()
):
    if x_api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    if idempotency_key in processed_requests:
        return processed_requests[idempotency_key]

    result = {
        "message_id": str(uuid4()),
        "status": "accepted",
        "provider": "MockMail"
    }

    processed_requests[idempotency_key] = result

    return result
```

***

## 29. Complete Notification Service

`notification_service/main.py`

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import uuid4
import httpx
import os


app = FastAPI(title="Notification Service")


EMAIL_PROVIDER_URL = "http://127.0.0.1:9000"

EMAIL_API_KEY = os.getenv(
    "EMAIL_API_KEY",
    "demo-secret-key"
)


class EmailNotification(BaseModel):
    recipient: str
    subject: str
    message: str


class EmailStatusEvent(BaseModel):
    message_id: str
    status: str


def build_provider_payload(
    notification: EmailNotification
):
    return {
        "to": notification.recipient,
        "subject": notification.subject,
        "content": notification.message
    }


@app.post("/notifications/email")
def send_notification(
    notification: EmailNotification
):
    payload = build_provider_payload(notification)

    idempotency_key = str(uuid4())

    headers = {
        "X-API-Key": EMAIL_API_KEY,
        "Idempotency-Key": idempotency_key
    }

    try:
        response = httpx.post(
            f"{EMAIL_PROVIDER_URL}/api/send",
            json=payload,
            headers=headers,
            timeout=5.0
        )

    except httpx.RequestError:
        raise HTTPException(
            status_code=503,
            detail="Email provider is unavailable"
        )

    if response.status_code == 401:
        raise HTTPException(
            status_code=502,
            detail="Email provider authentication failed"
        )

    if response.status_code == 429:
        raise HTTPException(
            status_code=503,
            detail="Email provider rate limit exceeded"
        )

    if response.status_code >= 500:
        raise HTTPException(
            status_code=502,
            detail="Email provider error"
        )

    if response.status_code != 200:
        raise HTTPException(
            status_code=400,
            detail="Email request rejected"
        )

    return response.json()


@app.post("/webhooks/email-status")
def receive_email_status(
    event: EmailStatusEvent
):
    print(
        f"Email {event.message_id}: "
        f"{event.status}"
    )

    return {
        "received": True
    }
```

***

## 30. Testing the Integration

Run the mock provider:

```bash
uvicorn main:app --reload --port 9000
```

Run Notification Service:

```bash
uvicorn main:app --reload --port 8000
```

Test:

```
POST /notifications/email
```

Request:

```json
{
    "recipient": "student@example.com",
    "subject": "Order Confirmation",
    "message": "Your order has been received."
}
```

Then test the webhook:

```
POST /webhooks/email-status
```

Request:

```json
{
    "message_id": "msg-123",
    "status": "delivered"
}
```

***

## 31. Practice Tasks

### Task 1 — Add Notification Types

Extend `EmailNotification` with:

```
notification_type
```

Possible values:

```
order
promotion
system
```

Include this information when processing the notification.

***

### Task 2 — Simulate Provider Failure

Modify the mock provider so that a particular email address:

```
fail@example.com
```

returns:

```
500 Internal Server Error
```

Verify that Notification Service correctly converts this into an appropriate error response.

***

### Task 3 — Simulate Rate Limiting

Modify the provider to return:

```
429 Too Many Requests
```

after a certain number of calls.

Observe how Notification Service handles the condition.

***

### Task 4 — Add Another Webhook Event

Support:

```
failed
```

in addition to:

```
delivered
```

Example:

```json
{
    "message_id": "msg-123",
    "status": "failed"
}
```

Process and display the event appropriately.

***

## 32. Key Takeaways

* Third-party services expose APIs that allow applications to use external capabilities.
* Applications should communicate with providers through well-defined HTTP APIs.
* Internal application models should not be unnecessarily coupled to provider-specific request formats.
* API keys and access tokens are commonly used for authentication.
* Sensitive credentials should be stored in configuration or environment variables.
* External API calls may fail because of network errors, timeouts, invalid credentials, rate limits, or provider errors.
* Idempotency helps prevent duplicate processing.
* Webhooks allow third-party providers to send events back to our application.
* Sandbox environments should be used for development and testing before switching to production.